1# Copyright 2024 The Android Open Source Project
2#
3# Licensed under the Apache License, Version 2.0 (the "License");
4# you may not use this file except in compliance with the License.
5# You may obtain a copy of the License at
6#
7#      http://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS,
11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12# See the License for the specific language governing permissions and
13# limitations under the License.
14
15import json
16import logging
17import os
18import os.path
19from pathlib import Path
20import re
21import subprocess
22import tempfile
23import time
24import multi_device_utils
25import yaml
26
27
28RESULT_KEY = 'result'
29RESULT_PASS = 'PASS'
30RESULT_FAIL = 'FAIL'
31CONFIG_FILE = os.path.join(os.getcwd(), 'config.yml')
32TESTS_DIR = os.path.join(os.getcwd(), 'tests')
33CTS_VERIFIER_PACKAGE_NAME = 'com.android.cts.verifier'
34MOBLY_TEST_SUMMARY_TXT_FILE = 'test_mobly_summary.txt'
35MULTI_DEVICE_TEST_ACTIVITY = (
36    'com.android.cts.verifier/.multidevice.MultiDeviceTestsActivity'
37)
38ACTION_HOST_TEST_RESULT = 'com.android.cts.verifier.ACTION_HOST_TEST_RESULT'
39EXTRA_VERSION = 'com.android.cts.verifier.extra.HOST_TEST_RESULT'
40ACTIVITY_START_WAIT = 2  # seconds
41
42
43def get_config_file_contents():
44  """Read the config file contents from a YML file.
45
46  Args: None
47
48  Returns:
49    config_file_contents: a dict read from config.yml
50  """
51  with open(CONFIG_FILE) as file:
52    config_file_contents = yaml.safe_load(file)
53  return config_file_contents
54
55
56def get_device_serial_number(config_file_contents):
57  """Returns the serial number of the dut devices.
58
59  Args:
60      config_file_contents: dict read from config.yml file.
61
62  Returns:
63      The serial numbers (str) or None if the device is not found.
64  """
65
66  device_serial_numbers = []
67  for _, testbed_data in config_file_contents.items():
68    for data_dict in testbed_data:
69      android_devices = data_dict.get('Controllers', {}).get(
70          'AndroidDevice', []
71      )
72
73      for device_dict in android_devices:
74        device_serial_numbers.append(device_dict.get('serial'))
75  return device_serial_numbers
76
77
78def report_result(device_id, results):
79  """Sends a pass/fail result to the device, via an intent.
80
81  Args:
82   device_id: the serial number of the device.
83   results: a dictionary contains all multi-device test names as key and
84     result/summary of current test run.
85  """
86  adb = f'adb -s {device_id}'
87
88  # Start MultiDeviceTestsActivity to receive test results
89  cmd = (
90      f'{adb} shell am start'
91      f' {MULTI_DEVICE_TEST_ACTIVITY} --activity-brought-to-front'
92  )
93  multi_device_utils.run(cmd)
94  time.sleep(ACTIVITY_START_WAIT)
95
96  json_results = json.dumps(results)
97  cmd = (
98      f'{adb} shell am broadcast -a {ACTION_HOST_TEST_RESULT} --es'
99      f" {EXTRA_VERSION} '{json_results}'"
100  )
101  if len(cmd) > 4095:
102    logging.info('Command string might be too long! len:%s', len(cmd))
103  multi_device_utils.run(cmd)
104
105
106def main():
107  """Run all Multi-device Mobly tests and collect results."""
108
109  logging.basicConfig(level=logging.INFO)
110  topdir = tempfile.mkdtemp(prefix='MultiDevice_')
111  subprocess.call(['chmod', 'g+rx', topdir])  # Add permissions
112  logging.info('Saving multi-device tests output files to: %s', topdir)
113
114  config_file_contents = get_config_file_contents()
115  device_ids = get_device_serial_number(config_file_contents)
116
117  test_results = {}
118  test_summary_file_list = []
119
120  # Run tests
121  for root, _, files in os.walk(TESTS_DIR):
122    for file in files:
123      if file.endswith('_test.py'):
124        test_name = os.path.splitext(file)[0]
125        test_file_path = os.path.join(root, file)
126        logging.info('Start running test: %s', test_name)
127        cmd = [
128            'python3',
129            test_file_path,  # Use the full path to the test file
130            '-c',
131            CONFIG_FILE,
132            '--testbed',
133            test_name,
134        ]
135        summary_file_path = os.path.join(topdir, MOBLY_TEST_SUMMARY_TXT_FILE)
136
137        test_completed = False
138        with open(summary_file_path, 'w+') as fp:
139          subprocess.run(cmd, stdout=fp, check=False)
140          fp.seek(0)
141          for line in fp:
142            if line.startswith('Test summary saved in'):
143              match = re.search(r'"(.*?)"', line)  # Get file path
144              if match:
145                test_summary = Path(match.group(1))
146                if test_summary.exists():
147                  test_summary_file_list.append(test_summary)
148                  test_completed = True
149                  break
150        if not test_completed:
151          logging.error('Failed to get test summary file path')
152        os.remove(os.path.join(topdir, MOBLY_TEST_SUMMARY_TXT_FILE))
153
154  # Parse test summary files
155  for test_summary_file in test_summary_file_list:
156    with open(test_summary_file) as file:
157      test_summary_content = yaml.safe_load_all(file)
158      for doc in test_summary_content:
159        if doc['Type'] == 'Record':
160          test_key = f"{doc['Test Class']}#{doc['Test Name']}"
161          result = (
162              RESULT_PASS if doc['Result'] in ('PASS', 'SKIP') else RESULT_FAIL
163          )
164          test_results.setdefault(test_key, {RESULT_KEY: result})
165
166  for device_id in device_ids:
167    report_result(device_id, test_results)
168
169  logging.info('Test execution completed. Results: %s', test_results)
170
171
172if __name__ == '__main__':
173  main()
174