• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1# Lint as: python3
2#
3# Copyright 2020, The Android Open Source Project
4#
5# Licensed under the Apache License, Version 2.0 (the "License");
6# you may not use this file except in compliance with the License.
7# You may obtain a copy of the License at
8#
9#     http://www.apache.org/licenses/LICENSE-2.0
10#
11# Unless required by applicable law or agreed to in writing, software
12# distributed under the License is distributed on an "AS IS" BASIS,
13# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14# See the License for the specific language governing permissions and
15# limitations under the License.
16"""Utilities for C-Suite integration tests."""
17
18import argparse
19import contextlib
20import logging
21import os
22import pathlib
23import shlex
24import shutil
25import stat
26import subprocess
27import sys
28import tempfile
29from typing import Sequence, Text
30import zipfile
31import csuite_test
32
33# Export symbols to reduce the number of imports tests have to list.
34TestCase = csuite_test.TestCase  # pylint: disable=invalid-name
35get_device_serial = csuite_test.get_device_serial
36
37# Keep any created temporary directories for debugging test failures. The
38# directories do not need explicit removal since they are created using the
39# system's temporary-file facility.
40_KEEP_TEMP_DIRS = False
41
42
43class CSuiteHarness(contextlib.AbstractContextManager):
44  """Interface class for interacting with the C-Suite harness.
45
46  WARNING: Explicitly clean up created instances or use as a context manager.
47  Not doing so will result in a ResourceWarning for the implicit cleanup which
48  confuses the TradeFed Python test output parser.
49  """
50
51  def __init__(self):
52    self._suite_dir = pathlib.Path(tempfile.mkdtemp(prefix='csuite'))
53    logging.debug('Created harness directory: %s', self._suite_dir)
54
55    with zipfile.ZipFile(_get_standalone_zip_path(), 'r') as f:
56      f.extractall(self._suite_dir)
57
58    # Add owner-execute permission on scripts since zip does not preserve them.
59    self._launcher_binary = self._suite_dir.joinpath(
60        'android-csuite/tools/csuite-tradefed')
61    _add_owner_exec_permission(self._launcher_binary)
62
63    self._testcases_dir = self._suite_dir.joinpath('android-csuite/testcases')
64
65  def __exit__(self, unused_type, unused_value, unused_traceback):
66    self.cleanup()
67
68  def cleanup(self):
69    if _KEEP_TEMP_DIRS:
70      return
71    shutil.rmtree(self._suite_dir, ignore_errors=True)
72
73
74  def run_and_wait(self, flags: Sequence[Text]) -> subprocess.CompletedProcess:
75    """Starts the Tradefed launcher and waits for it to complete."""
76
77    env = os.environ.copy()
78
79    # Unset environment variables that would cause the script to think it's in a
80    # build tree.
81    env.pop('ANDROID_BUILD_TOP', None)
82    env.pop('ANDROID_HOST_OUT', None)
83
84    # Unset environment variables that would cause TradeFed to find test configs
85    # other than the ones created by the test.
86    env.pop('ANDROID_HOST_OUT_TESTCASES', None)
87    env.pop('ANDROID_TARGET_OUT_TESTCASES', None)
88
89    # Unset environment variables that might cause the suite to pick up a
90    # connected device that wasn't explicitly specified.
91    env.pop('ANDROID_SERIAL', None)
92
93    # Unset environment variables that might cause the TradeFed to load classes
94    # that weren't included in the standalone suite zip.
95    env.pop('TF_GLOBAL_CONFIG', None)
96
97    # Set the environment variable that TradeFed requires to find test modules.
98    env['ANDROID_TARGET_OUT_TESTCASES'] = self._testcases_dir
99    jdk17_path = '/jdk/jdk17/linux-x86'
100    if os.path.isdir(jdk17_path):
101      env['JAVA_HOME'] = jdk17_path
102      java_path = jdk17_path + '/bin'
103      env['PATH'] = java_path + ':' + env['PATH']
104
105    return _run_command([self._launcher_binary] + flags, env=env)
106
107
108class PackageRepository(contextlib.AbstractContextManager):
109  """A file-system based APK repository for use in tests.
110
111  WARNING: Explicitly clean up created instances or use as a context manager.
112  Not doing so will result in a ResourceWarning for the implicit cleanup which
113  confuses the TradeFed Python test output parser.
114  """
115
116  def __init__(self):
117    self._root_dir = pathlib.Path(tempfile.mkdtemp(prefix='csuite_apk_dir'))
118    logging.info('Created repository directory: %s', self._root_dir)
119
120  def __exit__(self, unused_type, unused_value, unused_traceback):
121    self.cleanup()
122
123  def cleanup(self):
124    if _KEEP_TEMP_DIRS:
125      return
126    shutil.rmtree(self._root_dir, ignore_errors=True)
127
128  def get_path(self) -> pathlib.Path:
129    """Returns the path to the repository's root directory."""
130    return self._root_dir
131
132  def add_package_apks(self, package_name: Text,
133                       apk_paths: Sequence[pathlib.Path]):
134    """Adds the provided package APKs to the repository."""
135    apk_dir = self._root_dir.joinpath(package_name)
136
137    # Raises if the directory already exists.
138    apk_dir.mkdir()
139    for f in apk_paths:
140      shutil.copy(f, apk_dir)
141
142
143class Adb:
144  """Encapsulates adb functionality to simplify usage in tests.
145
146  Most methods in this class raise an exception if they fail to execute. This
147  behavior can be overridden by using the check parameter.
148  """
149
150  def __init__(self,
151               adb_binary_path: pathlib.Path = None,
152               device_serial: Text = None):
153    self._args = [adb_binary_path or 'adb']
154
155    device_serial = device_serial or get_device_serial()
156    if device_serial:
157      self._args.extend(['-s', device_serial])
158
159  def shell(self,
160            args: Sequence[Text],
161            check: bool = None) -> subprocess.CompletedProcess:
162    """Runs an adb shell command and waits for it to complete.
163
164    Note that the exit code of the returned object corresponds to that of
165    the adb command and not the command executed in the shell.
166
167    Args:
168      args: a sequence of program arguments to pass to the shell.
169      check: whether to raise if the process terminates with a non-zero exit
170        code.
171
172    Returns:
173      An object representing a process that has finished and that can be
174      queried.
175    """
176    return self.run(['shell'] + args, check)
177
178  def run(self,
179          args: Sequence[Text],
180          check: bool = None) -> subprocess.CompletedProcess:
181    """Runs an adb command and waits for it to complete."""
182    return _run_command(self._args + args, check=check)
183
184  def uninstall(self, package_name: Text, check: bool = None):
185    """Uninstalls the specified package."""
186    self.run(['uninstall', package_name], check=check)
187
188  def list_packages(self) -> Sequence[Text]:
189    """Lists packages installed on the device."""
190    p = self.shell(['pm', 'list', 'packages'])
191    return [l.split(':')[1] for l in p.stdout.splitlines()]
192
193
194def _run_command(args, check=False, **kwargs) -> subprocess.CompletedProcess:
195  """A wrapper for subprocess.run that overrides defaults and adds logging."""
196  env = kwargs.get('env', {})
197
198  # Log the command-line for debugging failed tests. Note that we convert
199  # tokens to strings for _shlex_join.
200  env_str = ['env', '-i'] + [f'{k}={v}' for k, v in env.items()]
201  args_str = [str(t) for t in args]
202
203  # Override some defaults. Note that 'check' deviates from this pattern to
204  # avoid getting warnings about using subprocess.run without an explicitly set
205  # `check` parameter.
206  kwargs.setdefault('capture_output', True)
207  kwargs.setdefault('universal_newlines', True)
208
209  logging.debug('Running command: %s', _shlex_join(env_str + args_str))
210
211  return subprocess.run(args, check=check, **kwargs)
212
213
214def _add_owner_exec_permission(path: pathlib.Path):
215  path.chmod(path.stat().st_mode | stat.S_IEXEC)
216
217
218def get_test_app_apks(app_module_name: Text) -> Sequence[pathlib.Path]:
219  """Returns a test app's apk file paths."""
220  return [_get_test_file(app_module_name + '.apk')]
221
222
223def _get_standalone_zip_path():
224  """Returns the suite standalone zip file's path."""
225  return _get_test_file('csuite-standalone.zip')
226
227
228def _get_test_file(name: Text) -> pathlib.Path:
229  test_dir = _get_test_dir()
230  test_file = test_dir.joinpath(name)
231
232  if not test_file.exists():
233    raise RuntimeError(f'Unable to find the file `{name}` in the test '
234                       'execution dir `{test_dir}`; are you missing a data '
235                       'dependency in the build module?')
236
237  return test_file
238
239
240def _shlex_join(split_command: Sequence[Text]) -> Text:
241  """Concatenate tokens and return a shell-escaped string."""
242  # This is an alternative to shlex.join that doesn't exist in Python versions
243  # < 3.8.
244  return ' '.join(shlex.quote(t) for t in split_command)
245
246
247def _get_test_dir() -> pathlib.Path:
248  return pathlib.Path(__file__).parent
249
250
251def main():
252  global _KEEP_TEMP_DIRS
253
254  parser = argparse.ArgumentParser(parents=[csuite_test.create_arg_parser()])
255  parser.add_argument(
256      '--log-level',
257      choices=['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'],
258      default='WARNING',
259      help='sets the logging level threshold')
260  parser.add_argument(
261      '--keep-temp-dirs',
262      type=bool,
263      help='keeps any created temporary directories for debugging failures')
264  args, unittest_argv = parser.parse_known_args(sys.argv)
265
266  _KEEP_TEMP_DIRS = args.keep_temp_dirs
267  logging.basicConfig(level=getattr(logging, args.log_level))
268
269  csuite_test.run_tests(args, unittest_argv)
270