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
18import shutil
19import tempfile
20import time
21
22import tzlocal
23from acts.controllers.android_device import SL4A_APK_NAME
24from acts.metrics.loggers.blackbox import BlackboxMappedMetricLogger
25from acts.test_utils.instrumentation import instrumentation_proto_parser \
26    as proto_parser
27from acts.test_utils.instrumentation.device.apps.app_installer import \
28    AppInstaller
29from acts.test_utils.instrumentation.device.command.adb_command_types import \
30    DeviceGServices
31from acts.test_utils.instrumentation.device.command.adb_command_types import \
32    DeviceSetprop
33from acts.test_utils.instrumentation.device.command.adb_command_types import \
34    DeviceSetting
35from acts.test_utils.instrumentation.device.command.adb_commands import common
36from acts.test_utils.instrumentation.device.command.adb_commands import goog
37from acts.test_utils.instrumentation.device.command.instrumentation_command_builder \
38    import DEFAULT_NOHUP_LOG
39from acts.test_utils.instrumentation.device.command.instrumentation_command_builder \
40    import InstrumentationTestCommandBuilder
41from acts.test_utils.instrumentation.instrumentation_base_test \
42    import InstrumentationBaseTest
43from acts.test_utils.instrumentation.instrumentation_base_test \
44    import InstrumentationTestError
45from acts.test_utils.instrumentation.instrumentation_proto_parser import \
46    DEFAULT_INST_LOG_DIR
47from acts.test_utils.instrumentation.power.power_metrics import Measurement
48from acts.test_utils.instrumentation.power.power_metrics import PowerMetrics
49from acts.test_utils.instrumentation.device.apps.permissions import PermissionsUtil
50
51from acts import asserts
52from acts import context
53
54ACCEPTANCE_THRESHOLD = 'acceptance_threshold'
55AUTOTESTER_LOG = 'autotester.log'
56DEFAULT_PUSH_FILE_TIMEOUT = 180
57DISCONNECT_USB_FILE = 'disconnectusb.log'
58POLLING_INTERVAL = 0.5
59
60
61class InstrumentationPowerTest(InstrumentationBaseTest):
62    """Instrumentation test for measuring and validating power metrics.
63
64    Params:
65        metric_logger: Blackbox metric logger used to store test metrics.
66        _instr_cmd_builder: Builder for the instrumentation command
67    """
68
69    def __init__(self, configs):
70        super().__init__(configs)
71
72        self.metric_logger = BlackboxMappedMetricLogger.for_test_case()
73        self._test_apk = None
74        self._sl4a_apk = None
75        self._instr_cmd_builder = None
76        self._power_metrics = None
77
78    def setup_class(self):
79        super().setup_class()
80        self.monsoon = self.monsoons[0]
81        self._setup_monsoon()
82
83    def setup_test(self):
84        """Test setup"""
85        super().setup_test()
86        self._prepare_device()
87        self._instr_cmd_builder = self.power_instrumentation_command_builder()
88        return True
89
90    def _prepare_device(self):
91        """Prepares the device for power testing."""
92        super()._prepare_device()
93        self._cleanup_test_files()
94        self._permissions_util = PermissionsUtil(
95            self.ad_dut,
96            self.get_file_from_config('permissions_apk'))
97        self._permissions_util.grant_all()
98        self._install_test_apk()
99
100    def _cleanup_device(self):
101        """Clean up device after power testing."""
102        if self._test_apk:
103            self._test_apk.uninstall()
104        self._permissions_util.close()
105        self._cleanup_test_files()
106
107    def base_device_configuration(self):
108        """Run the base setup commands for power testing."""
109        self.log.info('Running base device setup commands.')
110
111        self.ad_dut.adb.ensure_root()
112        self.adb_run(common.dismiss_keyguard)
113        self.ad_dut.ensure_screen_on()
114
115        # Test harness flag
116        self.adb_run(common.test_harness.toggle(True))
117
118        # Calling
119        self.adb_run(common.disable_dialing.toggle(True))
120
121        # Screen
122        self.adb_run(common.screen_always_on.toggle(True))
123        self.adb_run(common.screen_adaptive_brightness.toggle(False))
124
125        brightness_level = None
126        if 'brightness_level' in self._instrumentation_config:
127            brightness_level = self._instrumentation_config['brightness_level']
128
129        if brightness_level is None:
130            raise ValueError('no brightness level defined (or left as None) '
131                             'and it is needed.')
132
133        self.adb_run(common.screen_brightness.set_value(brightness_level))
134        self.adb_run(common.screen_timeout_ms.set_value(1800000))
135        self.adb_run(common.notification_led.toggle(False))
136        self.adb_run(common.screensaver.toggle(False))
137        self.adb_run(common.wake_gesture.toggle(False))
138        self.adb_run(common.doze_mode.toggle(False))
139        self.adb_run(common.doze_always_on.toggle(False))
140
141        # Sensors
142        self.adb_run(common.auto_rotate.toggle(False))
143        self.adb_run(common.disable_sensors)
144        self.adb_run(common.ambient_eq.toggle(False))
145
146        if self.file_exists(common.MOISTURE_DETECTION_SETTING_FILE):
147            self.adb_run(common.disable_moisture_detection)
148
149        # Time
150        self.adb_run(common.auto_time.toggle(False))
151        self.adb_run(common.auto_timezone.toggle(False))
152        self.adb_run(common.timezone.set_value(str(tzlocal.get_localzone())))
153
154        # Location
155        self.adb_run(common.location_gps.toggle(False))
156        self.adb_run(common.location_network.toggle(False))
157
158        # Power
159        self.adb_run(common.battery_saver_mode.toggle(False))
160        self.adb_run(common.battery_saver_trigger.set_value(0))
161        self.adb_run(common.enable_full_batterystats_history)
162        self.adb_run(common.disable_doze)
163
164        # Camera
165        self.adb_run(DeviceSetprop(
166            'camera.optbar.hdr', 'true', 'false').toggle(True))
167
168        # Gestures
169        gestures = {
170            'doze_pulse_on_pick_up': False,
171            'doze_pulse_on_double_tap': False,
172            'camera_double_tap_power_gesture_disabled': True,
173            'camera_double_twist_to_flip_enabled': False,
174            'assist_gesture_enabled': False,
175            'assist_gesture_silence_alerts_enabled': False,
176            'assist_gesture_wake_enabled': False,
177            'system_navigation_keys_enabled': False,
178            'camera_lift_trigger_enabled': False,
179            'doze_always_on': False,
180            'aware_enabled': False,
181            'doze_wake_screen_gesture': False,
182            'skip_gesture': False,
183            'silence_gesture': False
184        }
185        self.adb_run(
186            [DeviceSetting(common.SECURE, k).toggle(v)
187             for k, v in gestures.items()])
188
189        # GServices
190        self.adb_run(goog.location_collection.toggle(False))
191        self.adb_run(goog.cast_broadcast.toggle(False))
192        self.adb_run(DeviceGServices(
193            'location:compact_log_enabled').toggle(True))
194        self.adb_run(DeviceGServices('gms:magictether:enable').toggle(False))
195        self.adb_run(DeviceGServices('ocr.cc_ocr_enabled').toggle(False))
196        self.adb_run(DeviceGServices(
197            'gms:phenotype:phenotype_flag:debug_bypass_phenotype').toggle(True))
198        self.adb_run(DeviceGServices(
199            'gms_icing_extension_download_enabled').toggle(False))
200
201        # Comms
202        self.adb_run(common.wifi.toggle(False))
203        self.adb_run(common.bluetooth.toggle(False))
204        self.adb_run(common.airplane_mode.toggle(True))
205
206        # Misc. Google features
207        self.adb_run(goog.disable_playstore)
208        self.adb_run(goog.disable_volta)
209        self.adb_run(goog.disable_chre)
210        self.adb_run(goog.disable_musiciq)
211        self.adb_run(goog.disable_hotword)
212
213        # Enable clock dump info
214        self.adb_run('echo 1 > /d/clk/debug_suspend')
215
216    def _setup_monsoon(self):
217        """Set up the Monsoon controller for this testclass/testcase."""
218        self.log.info('Setting up Monsoon %s' % self.monsoon.serial)
219        monsoon_config = self._get_merged_config('Monsoon')
220        self._monsoon_voltage = monsoon_config.get_numeric('voltage', 4.2)
221        self.monsoon.set_voltage_safe(self._monsoon_voltage)
222        if 'max_current' in monsoon_config:
223            self.monsoon.set_max_current(
224                monsoon_config.get_numeric('max_current'))
225
226        self.monsoon.usb('on')
227        self.monsoon.set_on_disconnect(self._on_disconnect)
228        self.monsoon.set_on_reconnect(self._on_reconnect)
229
230        self._disconnect_usb_timeout = monsoon_config.get_numeric(
231            'usb_disconnection_timeout', 240)
232
233        self._measurement_args = dict(
234            duration=monsoon_config.get_numeric('duration'),
235            hz=monsoon_config.get_numeric('frequency'),
236            measure_after_seconds=monsoon_config.get_numeric('delay')
237        )
238
239    def _on_disconnect(self):
240        """Callback invoked by device disconnection from the Monsoon."""
241        self.ad_dut.log.info('Disconnecting device.')
242        self.ad_dut.stop_services()
243        # Uninstall SL4A
244        self._sl4a_apk = AppInstaller.pull_from_device(
245            self.ad_dut, SL4A_APK_NAME, tempfile.mkdtemp(prefix='sl4a'))
246        self._sl4a_apk.uninstall()
247        time.sleep(1)
248
249    def _on_reconnect(self):
250        """Callback invoked by device reconnection to the Monsoon"""
251        # Reinstall SL4A
252        if not self.ad_dut.is_sl4a_installed() and self._sl4a_apk:
253            self._sl4a_apk.install()
254            shutil.rmtree(os.path.dirname(self._sl4a_apk.apk_path))
255            self._sl4a_apk = None
256        self.ad_dut.start_services()
257        # Release wake lock to put device into sleep.
258        self.ad_dut.droid.goToSleepNow()
259        self.ad_dut.log.info('Device reconnected.')
260
261    def _install_test_apk(self):
262        """Installs test apk on the device."""
263        test_apk_file = self.get_file_from_config('test_apk')
264        self._test_apk = AppInstaller(self.ad_dut, test_apk_file)
265        self._test_apk.install('-g')
266        if not self._test_apk.is_installed():
267            raise InstrumentationTestError('Failed to install test APK.')
268
269    def _cleanup_test_files(self):
270        """Remove test-generated files from the device."""
271        self.ad_dut.log.info('Cleaning up test generated files.')
272        for file_name in [DISCONNECT_USB_FILE, DEFAULT_INST_LOG_DIR,
273                          DEFAULT_NOHUP_LOG, AUTOTESTER_LOG]:
274            path = os.path.join(self.ad_dut.external_storage_path, file_name)
275            self.adb_run('rm -rf %s' % path)
276
277    def trigger_scan_on_external_storage(self):
278        cmd = 'am broadcast -a android.intent.action.MEDIA_MOUNTED '
279        cmd = cmd + '-d file://%s ' % self.ad_dut.external_storage_path
280        cmd = cmd + '--receiver-include-background'
281        return self.adb_run(cmd)
282
283    def file_exists(self, file_path):
284        cmd = '(test -f %s && echo yes) || echo no' % file_path
285        result = self.adb_run(cmd)
286        if result[cmd] == 'yes':
287            return True
288        elif result[cmd] == 'no':
289            return False
290        raise ValueError('Couldn\'t determine if %s exists. '
291                         'Expected yes/no, got %s' % (file_path, result[cmd]))
292
293    def push_to_external_storage(self, file_path, dest=None,
294        timeout=DEFAULT_PUSH_FILE_TIMEOUT):
295        """Pushes a file to {$EXTERNAL_STORAGE} and returns its final location.
296
297        Args:
298            file_path: The file to be pushed.
299            dest: Where within {$EXTERNAL_STORAGE} it should be pushed.
300            timeout: Float number of seconds to wait for the file to be pushed.
301
302        Returns: The absolute path where the file was pushed.
303        """
304        if dest is None:
305            dest = os.path.basename(file_path)
306
307        dest_path = os.path.join(self.ad_dut.external_storage_path, dest)
308        self.log.info('clearing %s before pushing %s' % (dest_path, file_path))
309        self.ad_dut.adb.shell('rm -rf %s', dest_path)
310        self.log.info('pushing file %s to %s' % (file_path, dest_path))
311        self.ad_dut.adb.push(file_path, dest_path, timeout=timeout)
312        return dest_path
313
314    # Test runtime utils
315
316    def power_instrumentation_command_builder(self):
317        """Return the default command builder for power tests"""
318        builder = InstrumentationTestCommandBuilder.default()
319        builder.set_manifest_package(self._test_apk.pkg_name)
320        builder.set_nohup()
321        return builder
322
323    def _wait_for_disconnect_signal(self):
324        """Poll the device for a disconnect USB signal file. This will indicate
325        to the Monsoon that the device is ready to be disconnected.
326        """
327        self.log.info('Waiting for USB disconnect signal')
328        disconnect_file = os.path.join(
329            self.ad_dut.external_storage_path, DISCONNECT_USB_FILE)
330        start_time = time.time()
331        while time.time() < start_time + self._disconnect_usb_timeout:
332            if self.ad_dut.adb.shell('ls %s' % disconnect_file):
333                return
334            time.sleep(POLLING_INTERVAL)
335        raise InstrumentationTestError('Timeout while waiting for USB '
336                                       'disconnect signal.')
337
338    def measure_power(self):
339        """Measures power consumption with the Monsoon. See monsoon_lib API for
340        details.
341        """
342        if not hasattr(self, '_measurement_args'):
343            raise InstrumentationTestError('Missing Monsoon measurement args.')
344
345        # Start measurement after receiving disconnect signal
346        self._wait_for_disconnect_signal()
347        power_data_path = os.path.join(
348            context.get_current_context().get_full_output_path(), 'power_data')
349        self.log.info('Starting Monsoon measurement.')
350        self.monsoon.usb('auto')
351        measure_start_time = time.time()
352        result = self.monsoon.measure_power(
353            **self._measurement_args, output_path=power_data_path)
354        self.monsoon.usb('on')
355        self.log.info('Monsoon measurement complete.')
356
357        # Gather relevant metrics from measurements
358        session = self.dump_instrumentation_result_proto()
359        self._power_metrics = PowerMetrics(self._monsoon_voltage,
360                                           start_time=measure_start_time)
361        self._power_metrics.generate_test_metrics(
362            PowerMetrics.import_raw_data(power_data_path),
363            proto_parser.get_test_timestamps(session))
364        self._log_metrics()
365        return result
366
367    def run_and_measure(self, instr_class, instr_method=None, req_params=None,
368        extra_params=None):
369        """Convenience method for setting up the instrumentation test command,
370        running it on the device, and starting the Monsoon measurement.
371
372        Args:
373            instr_class: Fully qualified name of the instrumentation test class
374            instr_method: Name of the instrumentation test method
375            req_params: List of required parameter names
376            extra_params: List of ad-hoc parameters to be passed defined as
377                tuples of size 2.
378
379        Returns: summary of Monsoon measurement
380        """
381        if instr_method:
382            self._instr_cmd_builder.add_test_method(instr_class, instr_method)
383        else:
384            self._instr_cmd_builder.add_test_class(instr_class)
385        params = {}
386        instr_call_config = self._get_merged_config('instrumentation_call')
387        # Add required parameters
388        for param_name in req_params or []:
389            params[param_name] = instr_call_config.get(
390                param_name, verify_fn=lambda x: x is not None,
391                failure_msg='%s is a required parameter.' % param_name)
392        # Add all other parameters
393        params.update(instr_call_config)
394        for name, value in params.items():
395            self._instr_cmd_builder.add_key_value_param(name, value)
396
397        if extra_params:
398            for name, value in extra_params:
399                self._instr_cmd_builder.add_key_value_param(name, value)
400
401        instr_cmd = self._instr_cmd_builder.build()
402        self.log.info('Running instrumentation call: %s' % instr_cmd)
403        self.adb_run_async(instr_cmd)
404        return self.measure_power()
405
406    def _log_metrics(self):
407        """Record the collected metrics with the metric logger."""
408        self.log.info('Obtained metrics summaries:')
409        for k, m in self._power_metrics.test_metrics.items():
410            self.log.info('%s %s' % (k, str(m.summary)))
411
412        for metric_name in PowerMetrics.ALL_METRICS:
413            for instr_test_name in self._power_metrics.test_metrics:
414                metric_value = getattr(
415                    self._power_metrics.test_metrics[instr_test_name],
416                    metric_name).value
417                # TODO: Refactor this into instr_test_name.metric_name
418                self.metric_logger.add_metric(
419                    '%s__%s' % (metric_name, instr_test_name), metric_value)
420
421    def validate_power_results(self, *instr_test_names):
422        """Compare power measurements with target values and set the test result
423        accordingly.
424
425        Args:
426            instr_test_names: Name(s) of the instrumentation test method.
427                If none specified, defaults to all test methods run.
428
429        Raises:
430            signals.TestFailure if one or more metrics do not satisfy threshold
431        """
432        summaries = {}
433        failure = False
434        all_thresholds = self._get_merged_config(ACCEPTANCE_THRESHOLD)
435
436        if not instr_test_names:
437            instr_test_names = all_thresholds.keys()
438
439        for instr_test_name in instr_test_names:
440            try:
441                test_metrics = self._power_metrics.test_metrics[instr_test_name]
442            except KeyError:
443                raise InstrumentationTestError(
444                    'Unable to find test method %s in instrumentation output. '
445                    'Check instrumentation call results in '
446                    'instrumentation_proto.txt.'
447                    % instr_test_name)
448
449            summaries[instr_test_name] = {}
450            test_thresholds = all_thresholds.get_config(instr_test_name)
451            for metric_name, metric in test_thresholds.items():
452                try:
453                    actual_result = getattr(test_metrics, metric_name)
454                except AttributeError:
455                    continue
456
457                if 'unit_type' not in metric or 'unit' not in metric:
458                    continue
459                unit_type = metric['unit_type']
460                unit = metric['unit']
461
462                lower_value = metric.get_numeric('lower_limit', float('-inf'))
463                upper_value = metric.get_numeric('upper_limit', float('inf'))
464                if 'expected_value' in metric and 'percent_deviation' in metric:
465                    expected_value = metric.get_numeric('expected_value')
466                    percent_deviation = metric.get_numeric('percent_deviation')
467                    lower_value = expected_value * (1 - percent_deviation / 100)
468                    upper_value = expected_value * (1 + percent_deviation / 100)
469
470                lower_bound = Measurement(lower_value, unit_type, unit)
471                upper_bound = Measurement(upper_value, unit_type, unit)
472                summary_entry = {
473                    'expected': '[%s, %s]' % (lower_bound, upper_bound),
474                    'actual': str(actual_result.to_unit(unit))
475                }
476                summaries[instr_test_name][metric_name] = summary_entry
477                if not lower_bound <= actual_result <= upper_bound:
478                    failure = True
479        self.log.info('Summary of measurements: %s' % summaries)
480        asserts.assert_false(
481            failure,
482            msg='One or more measurements do not meet the specified criteria',
483            extras=summaries)
484        asserts.explicit_pass(
485            msg='All measurements meet the criteria',
486            extras=summaries)
487