systrace/catapult/devil/devil/android/tools/device_monitor_test.py

#!/usr/bin/env python
# Copyright 2015 The Chromium Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

import os
import sys
import unittest

import six

if __name__ == '__main__':
  sys.path.append(
      os.path.abspath(
          os.path.join(os.path.dirname(__file__), '..', '..', '..')))

from devil import devil_env
from devil.android import device_errors
from devil.android import device_utils
from devil.android.tools import device_monitor

with devil_env.SysPath(devil_env.PYMOCK_PATH):
  import mock  # pylint: disable=import-error


class DeviceMonitorTest(unittest.TestCase):
  def setUp(self):
    self.device = mock.Mock(
        spec=device_utils.DeviceUtils,
        serial='device_cereal',
        build_id='abc123',
        build_product='clownfish',
        GetIMEI=lambda: '123456789')
    self.file_contents = {
        '/proc/meminfo':
            """
                         MemTotal:        1234567 kB
                         MemFree:         1000000 kB
                         MemUsed:          234567 kB
                         """,
        '/sys/class/thermal/thermal_zone0/type':
            'CPU-therm',
        '/sys/class/thermal/thermal_zone0/temp':
            '30',
        '/proc/uptime':
            '12345 99999',
    }
    self.device.ReadFile = mock.MagicMock(
        side_effect=lambda file_name: self.file_contents[file_name])

    self.device.ListProcesses.return_value = ['p1', 'p2', 'p3', 'p4', 'p5']

    self.cmd_outputs = {
        'grep': ['/sys/class/thermal/thermal_zone0/type'],
    }

    def mock_run_shell(cmd, **_kwargs):
      args = cmd.split() if isinstance(cmd, six.string_types) else cmd
      try:
        return self.cmd_outputs[args[0]]
      except KeyError:
        raise device_errors.AdbShellCommandFailedError(cmd, None, None)

    self.device.RunShellCommand = mock.MagicMock(side_effect=mock_run_shell)

    self.battery = mock.Mock()
    self.battery.GetBatteryInfo = mock.MagicMock(return_value={
        'level': '80',
        'temperature': '123'
    })

    self.expected_status = {
        'device_cereal': {
            'processes': 5,
            'temp': {
                'CPU-therm': 30.0
            },
            'battery': {
                'temperature': 123,
                'level': 80
            },
            'uptime': 12345.0,
            'mem': {
                'total': 1234567,
                'free': 1000000
            },
            'build': {
                'build.id': 'abc123',
                'product.device': 'clownfish',
            },
            'imei': '123456789',
            'state': 'available',
        }
    }

  @mock.patch('devil.android.battery_utils.BatteryUtils')
  @mock.patch('devil.android.device_utils.DeviceUtils.HealthyDevices')
  def test_getStats(self, get_devices, get_battery):
    get_devices.return_value = [self.device]
    get_battery.return_value = self.battery

    status = device_monitor.get_all_status(None)
    self.assertEquals(self.expected_status, status['devices'])

  @mock.patch('devil.android.battery_utils.BatteryUtils')
  @mock.patch('devil.android.device_utils.DeviceUtils.HealthyDevices')
  def test_getStatsNoBattery(self, get_devices, get_battery):
    get_devices.return_value = [self.device]
    get_battery.return_value = self.battery
    broken_battery_info = mock.Mock()
    broken_battery_info.GetBatteryInfo = mock.MagicMock(return_value={
        'level': '-1',
        'temperature': 'not_a_number'
    })
    get_battery.return_value = broken_battery_info

    # Should be same status dict but without battery stats.
    expected_status_no_battery = self.expected_status.copy()
    expected_status_no_battery['device_cereal'].pop('battery')

    status = device_monitor.get_all_status(None)
    self.assertEquals(expected_status_no_battery, status['devices'])

  @mock.patch('devil.android.battery_utils.BatteryUtils')
  @mock.patch('devil.android.device_utils.DeviceUtils.HealthyDevices')
  def test_getStatsNoPs(self, get_devices, get_battery):
    get_devices.return_value = [self.device]
    get_battery.return_value = self.battery
    # Throw exception when listing processes.
    self.device.ListProcesses.side_effect = device_errors.AdbCommandFailedError(
        ['ps'], 'something failed', 1)

    # Should be same status dict but without process stats.
    expected_status_no_ps = self.expected_status.copy()
    expected_status_no_ps['device_cereal'].pop('processes')

    status = device_monitor.get_all_status(None)
    self.assertEquals(expected_status_no_ps, status['devices'])

  @mock.patch('devil.android.battery_utils.BatteryUtils')
  @mock.patch('devil.android.device_utils.DeviceUtils.HealthyDevices')
  def test_getStatsNoSensors(self, get_devices, get_battery):
    get_devices.return_value = [self.device]
    get_battery.return_value = self.battery
    del self.cmd_outputs['grep']  # Throw exception on run shell grep command.

    # Should be same status dict but without temp stats.
    expected_status_no_temp = self.expected_status.copy()
    expected_status_no_temp['device_cereal'].pop('temp')

    status = device_monitor.get_all_status(None)
    self.assertEquals(expected_status_no_temp, status['devices'])

  @mock.patch('devil.android.battery_utils.BatteryUtils')
  @mock.patch('devil.android.device_utils.DeviceUtils.HealthyDevices')
  def test_getStatsWithDenylist(self, get_devices, get_battery):
    get_devices.return_value = [self.device]
    get_battery.return_value = self.battery
    denylist = mock.Mock()
    denylist.Read = mock.MagicMock(
        return_value={'bad_device': {
            'reason': 'offline'
        }})

    # Should be same status dict but with extra denylisted device.
    expected_status = self.expected_status.copy()
    expected_status['bad_device'] = {'state': 'offline'}

    status = device_monitor.get_all_status(denylist)
    self.assertEquals(expected_status, status['devices'])

  @mock.patch('devil.android.battery_utils.BatteryUtils')
  @mock.patch('devil.android.device_utils.DeviceUtils.HealthyDevices')
  def test_brokenTempValue(self, get_devices, get_battery):
    self.file_contents['/sys/class/thermal/thermal_zone0/temp'] = 'n0t a numb3r'
    get_devices.return_value = [self.device]
    get_battery.return_value = self.battery

    expected_status_no_temp = self.expected_status.copy()
    expected_status_no_temp['device_cereal'].pop('temp')

    status = device_monitor.get_all_status(None)
    self.assertEquals(self.expected_status, status['devices'])


if __name__ == '__main__':
  sys.exit(unittest.main())