• 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
100    return _run_command([self._launcher_binary] + flags, env=env)
101
102
103class PackageRepository(contextlib.AbstractContextManager):
104  """A file-system based APK repository for use in tests.
105
106  WARNING: Explicitly clean up created instances or use as a context manager.
107  Not doing so will result in a ResourceWarning for the implicit cleanup which
108  confuses the TradeFed Python test output parser.
109  """
110
111  def __init__(self):
112    self._root_dir = pathlib.Path(tempfile.mkdtemp(prefix='csuite_apk_dir'))
113    logging.info('Created repository directory: %s', self._root_dir)
114
115  def __exit__(self, unused_type, unused_value, unused_traceback):
116    self.cleanup()
117
118  def cleanup(self):
119    if _KEEP_TEMP_DIRS:
120      return
121    shutil.rmtree(self._root_dir, ignore_errors=True)
122
123  def get_path(self) -> pathlib.Path:
124    """Returns the path to the repository's root directory."""
125    return self._root_dir
126
127  def add_package_apks(self, package_name: Text,
128                       apk_paths: Sequence[pathlib.Path]):
129    """Adds the provided package APKs to the repository."""
130    apk_dir = self._root_dir.joinpath(package_name)
131
132    # Raises if the directory already exists.
133    apk_dir.mkdir()
134    for f in apk_paths:
135      shutil.copy(f, apk_dir)
136
137
138class Adb:
139  """Encapsulates adb functionality to simplify usage in tests.
140
141  Most methods in this class raise an exception if they fail to execute. This
142  behavior can be overridden by using the check parameter.
143  """
144
145  def __init__(self,
146               adb_binary_path: pathlib.Path = None,
147               device_serial: Text = None):
148    self._args = [adb_binary_path or 'adb']
149
150    device_serial = device_serial or get_device_serial()
151    if device_serial:
152      self._args.extend(['-s', device_serial])
153
154  def shell(self,
155            args: Sequence[Text],
156            check: bool = None) -> subprocess.CompletedProcess:
157    """Runs an adb shell command and waits for it to complete.
158
159    Note that the exit code of the returned object corresponds to that of
160    the adb command and not the command executed in the shell.
161
162    Args:
163      args: a sequence of program arguments to pass to the shell.
164      check: whether to raise if the process terminates with a non-zero exit
165        code.
166
167    Returns:
168      An object representing a process that has finished and that can be
169      queried.
170    """
171    return self.run(['shell'] + args, check)
172
173  def run(self,
174          args: Sequence[Text],
175          check: bool = None) -> subprocess.CompletedProcess:
176    """Runs an adb command and waits for it to complete."""
177    return _run_command(self._args + args, check=check)
178
179  def uninstall(self, package_name: Text, check: bool = None):
180    """Uninstalls the specified package."""
181    self.run(['uninstall', package_name], check=check)
182
183  def list_packages(self) -> Sequence[Text]:
184    """Lists packages installed on the device."""
185    p = self.shell(['pm', 'list', 'packages'])
186    return [l.split(':')[1] for l in p.stdout.splitlines()]
187
188
189def _run_command(args, check=False, **kwargs) -> subprocess.CompletedProcess:
190  """A wrapper for subprocess.run that overrides defaults and adds logging."""
191  env = kwargs.get('env', {})
192
193  # Log the command-line for debugging failed tests. Note that we convert
194  # tokens to strings for _shlex_join.
195  env_str = ['env', '-i'] + ['%s=%s' % (k, v) for k, v in env.items()]
196  args_str = [str(t) for t in args]
197
198  # Override some defaults. Note that 'check' deviates from this pattern to
199  # avoid getting warnings about using subprocess.run without an explicitly set
200  # `check` parameter.
201  kwargs.setdefault('capture_output', True)
202  kwargs.setdefault('universal_newlines', True)
203
204  logging.debug('Running command: %s', _shlex_join(env_str + args_str))
205
206  return subprocess.run(args, check=check, **kwargs)
207
208
209def _add_owner_exec_permission(path: pathlib.Path):
210  path.chmod(path.stat().st_mode | stat.S_IEXEC)
211
212
213def get_test_app_apks(app_module_name: Text) -> Sequence[pathlib.Path]:
214  """Returns a test app's apk file paths."""
215  return [_get_test_file(app_module_name + '.apk')]
216
217
218def _get_standalone_zip_path():
219  """Returns the suite standalone zip file's path."""
220  return _get_test_file('csuite-standalone.zip')
221
222
223def _get_test_file(name: Text) -> pathlib.Path:
224  test_dir = _get_test_dir()
225  test_file = test_dir.joinpath(name)
226
227  if not test_file.exists():
228    raise RuntimeError('Unable to find the file `%s` in the test execution dir '
229                       '`%s`; are you missing a data dependency in the build '
230                       'module?' % (name, test_dir))
231
232  return test_file
233
234
235def _shlex_join(split_command: Sequence[Text]) -> Text:
236  """Concatenate tokens and return a shell-escaped string."""
237  # This is an alternative to shlex.join that doesn't exist in Python versions
238  # < 3.8.
239  return ' '.join(shlex.quote(t) for t in split_command)
240
241
242def _get_test_dir() -> pathlib.Path:
243  return pathlib.Path(__file__).parent
244
245
246def main():
247  global _KEEP_TEMP_DIRS
248
249  parser = argparse.ArgumentParser(parents=[csuite_test.create_arg_parser()])
250  parser.add_argument(
251      '--log-level',
252      choices=['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'],
253      default='WARNING',
254      help='sets the logging level threshold')
255  parser.add_argument(
256      '--keep-temp-dirs',
257      type=bool,
258      help='keeps any created temporary directories for debugging failures')
259  args, unittest_argv = parser.parse_known_args(sys.argv)
260
261  _KEEP_TEMP_DIRS = args.keep_temp_dirs
262  logging.basicConfig(level=getattr(logging, args.log_level))
263
264  csuite_test.run_tests(args, unittest_argv)
265