1#!/usr/bin/env python3
2#
3#   Copyright 2019 - 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
17import os
18
19import yaml
20from acts.keys import Config
21from acts.test_utils.instrumentation import instrumentation_proto_parser \
22    as proto_parser
23from acts.test_utils.instrumentation.config_wrapper import ConfigWrapper
24from acts.test_utils.instrumentation.device.command.adb_commands import common
25
26from acts import base_test
27from acts import context
28from acts import utils
29
30RESOLVE_FILE_MARKER = 'FILE'
31FILE_NOT_FOUND = 'File is missing from ACTS config'
32DEFAULT_INSTRUMENTATION_CONFIG_FILE = 'instrumentation_config.yaml'
33
34
35class InstrumentationTestError(Exception):
36    pass
37
38
39class InstrumentationBaseTest(base_test.BaseTestClass):
40    """Base class for tests based on am instrument."""
41
42    def __init__(self, configs):
43        """Initialize an InstrumentationBaseTest
44
45        Args:
46            configs: Dict representing the test configuration
47        """
48        super().__init__(configs)
49        # Take instrumentation config path directly from ACTS config if found,
50        # otherwise try to find the instrumentation config in the same directory
51        # as the ACTS config
52        instrumentation_config_path = ''
53        if 'instrumentation_config' in self.user_params:
54            instrumentation_config_path = (
55                self.user_params['instrumentation_config'][0])
56        elif Config.key_config_path.value in self.user_params:
57            instrumentation_config_path = os.path.join(
58                self.user_params[Config.key_config_path.value],
59                DEFAULT_INSTRUMENTATION_CONFIG_FILE)
60        self._instrumentation_config = ConfigWrapper()
61        if os.path.exists(instrumentation_config_path):
62            self._instrumentation_config = self._load_instrumentation_config(
63                instrumentation_config_path)
64            self._class_config = self._instrumentation_config.get_config(
65                self.__class__.__name__)
66        else:
67            self.log.warning(
68                'Instrumentation config file %s does not exist' %
69                instrumentation_config_path)
70
71    def _load_instrumentation_config(self, path):
72        """Load the instrumentation config file into an
73        InstrumentationConfigWrapper object.
74
75        Args:
76            path: Path to the instrumentation config file.
77
78        Returns: The loaded instrumentation config as an
79        InstrumentationConfigWrapper
80        """
81        try:
82            with open(path, mode='r', encoding='utf-8') as f:
83                config_dict = yaml.safe_load(f)
84        except Exception as e:
85            raise InstrumentationTestError(
86                'Cannot open or parse instrumentation config file %s'
87                % path) from e
88
89        # Write out a copy of the instrumentation config
90        with open(os.path.join(
91                self.log_path, 'instrumentation_config.yaml'),
92                  mode='w', encoding='utf-8') as f:
93            yaml.safe_dump(config_dict, f)
94
95        return ConfigWrapper(config_dict)
96
97    def setup_class(self):
98        """Class setup"""
99        self.ad_dut = self.android_devices[0]
100
101    def teardown_test(self):
102        """Test teardown. Takes bugreport and cleans up device."""
103        self._ad_take_bugreport(self.ad_dut, 'teardown_class',
104                                utils.get_current_epoch_time())
105        self._cleanup_device()
106
107    def _prepare_device(self):
108        """Prepares the device for testing."""
109        pass
110
111    def _cleanup_device(self):
112        """Clean up device after test completion."""
113        pass
114
115    def _get_merged_config(self, config_name):
116        """Takes the configs with config_name from the base, testclass, and
117        testcase levels and merges them together. When the same parameter is
118        defined in different contexts, the value from the most specific context
119        is taken.
120
121        Example:
122            self._instrumentation_config = {
123                'sample_config': {
124                    'val_a': 5,
125                    'val_b': 7
126                },
127                'ActsTestClass': {
128                    'sample_config': {
129                        'val_b': 3,
130                        'val_c': 6
131                    },
132                    'acts_test_case': {
133                        'sample_config': {
134                            'val_c': 10,
135                            'val_d': 2
136                        }
137                    }
138                }
139            }
140
141            self._get_merged_config('sample_config') returns
142            {
143                'val_a': 5,
144                'val_b': 3,
145                'val_c': 10,
146                'val_d': 2
147            }
148
149        Args:
150            config_name: Name of the config to fetch
151        Returns: The merged config, as a ConfigWrapper
152        """
153        merged_config = self._instrumentation_config.get_config(
154            config_name)
155        merged_config.update(self._class_config.get_config(config_name))
156        if self.current_test_name:
157            case_config = self._class_config.get_config(self.current_test_name)
158            merged_config.update(case_config.get_config(config_name))
159        return merged_config
160
161    def get_files_from_config(self, config_key):
162        """Get a list of file paths on host from self.user_params with the
163        given key. Verifies that each file exists.
164
165        Args:
166            config_key: Key in which the files are found.
167
168        Returns: list of str file paths
169        """
170        if config_key not in self.user_params:
171            raise InstrumentationTestError(
172                'Cannot get files for key "%s": Key missing from config.'
173                % config_key)
174        files = self.user_params[config_key]
175        for f in files:
176            if not os.path.exists(f):
177                raise InstrumentationTestError(
178                    'Cannot get files for key "%s": No file exists for %s.' %
179                    (config_key, f))
180        return files
181
182    def get_file_from_config(self, config_key):
183        """Get a single file path on host from self.user_params with the given
184        key. See get_files_from_config for details.
185        """
186        return self.get_files_from_config(config_key)[-1]
187
188    def adb_run(self, cmds):
189        """Run the specified command, or list of commands, with the ADB shell.
190
191        Args:
192            cmds: A string or list of strings representing ADB shell command(s)
193
194        Returns: dict mapping command to resulting stdout
195        """
196        if isinstance(cmds, str):
197            cmds = [cmds]
198        out = {}
199        for cmd in cmds:
200            out[cmd] = self.ad_dut.adb.shell(cmd)
201        return out
202
203    def adb_run_async(self, cmds):
204        """Run the specified command, or list of commands, with the ADB shell.
205        (async)
206
207        Args:
208            cmds: A string or list of strings representing ADB shell command(s)
209
210        Returns: dict mapping command to resulting subprocess.Popen object
211        """
212        if isinstance(cmds, str):
213            cmds = [cmds]
214        procs = {}
215        for cmd in cmds:
216            procs[cmd] = self.ad_dut.adb.shell_nb(cmd)
217        return procs
218
219    def dump_instrumentation_result_proto(self):
220        """Dump the instrumentation result proto as a human-readable txt file
221        in the log directory.
222
223        Returns: The parsed instrumentation_data_pb2.Session
224        """
225        session = proto_parser.get_session_from_device(self.ad_dut)
226        proto_txt_path = os.path.join(
227            context.get_current_context().get_full_output_path(),
228            'instrumentation_proto.txt')
229        with open(proto_txt_path, 'w') as f:
230            f.write(str(session))
231        return session
232
233    # Basic setup methods
234
235    def mode_airplane(self):
236        """Mode for turning on airplane mode only."""
237        self.log.info('Enabling airplane mode.')
238        self.adb_run(common.airplane_mode.toggle(True))
239        self.adb_run(common.auto_time.toggle(False))
240        self.adb_run(common.auto_timezone.toggle(False))
241        self.adb_run(common.location_gps.toggle(False))
242        self.adb_run(common.location_network.toggle(False))
243        self.adb_run(common.wifi.toggle(False))
244        self.adb_run(common.bluetooth.toggle(False))
245
246    def mode_wifi(self):
247        """Mode for turning on airplane mode and wifi."""
248        self.log.info('Enabling airplane mode and wifi.')
249        self.adb_run(common.airplane_mode.toggle(True))
250        self.adb_run(common.location_gps.toggle(False))
251        self.adb_run(common.location_network.toggle(False))
252        self.adb_run(common.wifi.toggle(True))
253        self.adb_run(common.bluetooth.toggle(False))
254
255    def mode_bluetooth(self):
256        """Mode for turning on airplane mode and bluetooth."""
257        self.log.info('Enabling airplane mode and bluetooth.')
258        self.adb_run(common.airplane_mode.toggle(True))
259        self.adb_run(common.auto_time.toggle(False))
260        self.adb_run(common.auto_timezone.toggle(False))
261        self.adb_run(common.location_gps.toggle(False))
262        self.adb_run(common.location_network.toggle(False))
263        self.adb_run(common.wifi.toggle(False))
264        self.adb_run(common.bluetooth.toggle(True))
265