xref: /aosp_15_r20/tools/asuite/atest/usb_speed_detect_unittest.py (revision c2e18aaa1096c836b086f94603d04f4eb9cf37f5)

# Copyright (C) 2024 The Android Open Source Project
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import subprocess
import unittest
from unittest import mock

from atest import atest_utils
from atest import usb_speed_detect as usb


class UsbIgnoredSpeedPatterns(unittest.TestCase):

  def _usb_speed_assert_no_warning(self, negotiated_speed, max_speed):
    """Parametrized test to verify whether a usb speed warning is printed."""
    warning = usb.verify_and_print_usb_speed_warning(
        device_ids=usb.DeviceIds('', '', '', '', ''),
        negotiated_speed=negotiated_speed,
        max_speed=max_speed,
    )

    self.assertFalse(warning)

  def test_verify_print_speed_unknown_speed_doesnt_print(self):
    self._usb_speed_assert_no_warning(0, 0)

  def test_verify_print_speed_low_speed_doesnt_print(self):
    self._usb_speed_assert_no_warning(480, 480)

  def test_verify_print_speed_expected_speed_doesnt_print(self):
    self._usb_speed_assert_no_warning(5000, 5000)

  def test_verify_print_speed_high_speed_doesnt_print(self):
    self._usb_speed_assert_no_warning(5000, 10000)


class UsbSpeedDetectTest(unittest.TestCase):

  def test_verify_print_speed_slow_speed_prints_warning(self):
    warning = usb.verify_and_print_usb_speed_warning(
        device_ids=usb.DeviceIds('', '', '', '', ''),
        negotiated_speed=480,
        max_speed=10000,
    )

    self.assertTrue(warning)


class UdcDriverPatterns(unittest.TestCase):

  def _udc_driver_response(
      self, attr_name: usb.UsbAttributeName, expected_response: int
  ):
    """Parametrized test for handling the responses from the usb driver."""

    speed = usb.get_udc_driver_usb_device_attribute_speed_value('', attr_name)

    self.assertEqual(speed, expected_response)

  @mock.patch('subprocess.check_output', return_value='not found')
  def test_udc_driver_unexpected_subprocess_response_returns_0(
      self, mock_output
  ):
    self._udc_driver_response(usb.UsbAttributeName.MAXIMUM_SPEED, 0)

  @mock.patch('subprocess.check_output', return_value='UNKNOWN')
  def test_udc_driver_unknown_speed_returns_0(self, mock_output):
    self._udc_driver_response(usb.UsbAttributeName.MAXIMUM_SPEED, 0)

  @mock.patch('subprocess.check_output', return_value='wireless')
  def test_udc_driver_irrelevant_speed_returns_0(self, mock_output):
    self._udc_driver_response(usb.UsbAttributeName.NEGOTIATED_SPEED, 0)

  @mock.patch('subprocess.check_output', return_value='high-speed')
  def test_udc_driver_high_speed_returns_numeric_speed(self, mock_output):
    self._udc_driver_response(usb.UsbAttributeName.MAXIMUM_SPEED, 480)

  @mock.patch('subprocess.check_output', return_value='high-speed\n')
  def test_udc_driver_high_speed_output_has_newline_returns_numeric_speed(
      self, mock_output
  ):
    self._udc_driver_response(usb.UsbAttributeName.MAXIMUM_SPEED, 480)

  @mock.patch('subprocess.check_output', return_value='super-speed')
  def test_udc_driver_super_speed_returns_numeric_speed(self, mock_output):
    self._udc_driver_response(usb.UsbAttributeName.MAXIMUM_SPEED, 5000)

  @mock.patch('subprocess.check_output', return_value='super-speed-plus')
  def test_udc_driver_super_speed_plus_returns_numeric_speed(self, mock_output):
    self._udc_driver_response(usb.UsbAttributeName.MAXIMUM_SPEED, 10000)


class DeviceIdentifierPatterns(unittest.TestCase):

  @mock.patch.object(atest_utils, 'has_command', return_value=True)
  @mock.patch.object(subprocess, 'check_output')
  def test_get_adb_device_identifiers_port_fwd_device_returns_address(
      self, mock_output, mock_utils
  ):
    def check_output_side_effect_port_fwd_device(*args, **kwargs):
      for arg in args:
        if 'ro.serialno' in arg:
          return 'SERIAL'
        if all(cmd_arg in ['adb', 'devices'] for cmd_arg in arg):
          return 'List of devices\nlocalhost:27030     device'
        if any(
            cmd_arg
            in {
                'ro.product.manufacturer',
                'ro.product.model',
                'ro.product.name',
            }
            for cmd_arg in arg
        ):
          return ''

    mock_output.side_effect = check_output_side_effect_port_fwd_device

    device_ids = usb.get_adb_device_identifiers()

    self.assertEqual(device_ids.address, 'localhost:27030')

  @mock.patch.object(atest_utils, 'has_command', return_value=True)
  @mock.patch.object(subprocess, 'check_output')
  def test_get_adb_device_identifiers_tcp_device_returns_address(
      self, mock_output, mock_utils
  ):
    def check_output_side_effect_tcp_device(*args, **kwargs):
      for arg in args:
        if 'ro.serialno' in arg:
          return 'SERIAL'
        if all(cmd_arg in ['adb', 'devices'] for cmd_arg in arg):
          return (
              '* daemon not running; starting now at tcp:1111\n * daemon '
              'started successfully\n List of devices\n33a832a820  device'
          )
        if any(
            # If check_output is called with any of ('model', 'name',
            # 'manufacturer', return an empty placeholder value.
            cmd_arg
            in {
                'ro.product.manufacturer',
                'ro.product.model',
                'ro.product.name',
            }
            for cmd_arg in arg
        ):
          return ''

    mock_output.side_effect = check_output_side_effect_tcp_device

    device_ids = usb.get_adb_device_identifiers()

    self.assertEqual(device_ids.address, '33a832a820')

  @mock.patch.object(atest_utils, 'has_command', return_value=True)
  @mock.patch.object(subprocess, 'check_output')
  def test_get_adb_device_identifiers_multiple_devices_returns_none(
      self, mock_output, mock_utils
  ):
    def check_output_side_effect_multiple_devices(*args, **kwargs):
      for arg in args:
        # When multiple devices are connected, ADB will display an error "adb:
        # more than one device/emulator" and no serial will be returned.
        if 'ro.serialno' in arg:
          return None

    mock_output.side_effect = check_output_side_effect_multiple_devices

    device_ids = usb.get_adb_device_identifiers()

    self.assertIsNone(device_ids)