| # Copyright 2023 Google LLC |
| # |
| # 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. |
| |
| """UI Automator controller in python layer. |
| |
| A python controller that can trigger mobly UI automator snippet to achieve some |
| automated UI operations on Android phones. |
| """ |
| import functools |
| import logging |
| import os |
| import re |
| from typing import Any, Callable |
| |
| from absl import app |
| from absl import flags |
| import inflection |
| from mobly.controllers import android_device |
| from mobly.controllers.android_device_lib import adb |
| from mobly.snippet import errors as snippet_errors |
| |
| from ui_automator import android_device as ad |
| from ui_automator import errors |
| from ui_automator import version |
| |
| |
| _GOOGLE_HOME_APP = { |
| 'id': 'gha', |
| 'name': 'Google Home Application (GHA)', |
| 'minVersion': '3.1.18.1', |
| 'minMatter': '231456000', |
| 'minThread': '231456000', |
| } |
| _MOBLY_SNIPPET_APK: str = 'com.chip.interop.moblysnippet' |
| _MOBLY_SNIPPET_APK_NAME: str = 'mbs' |
| _COMMISSIONING_FLAG_USAGE_GUIDE = ( |
| 'Use --commission {DeviceName},{PairingCode},{GHARoom} to commission a' |
| ' device to google fabric on GHA.' |
| ) |
| _REGRESSION_TESTS_FLAG_USAGE_GUIDE = ( |
| 'Use --regtest to run regression tests infinitely. Add --repeat' |
| ' <repeat-times> to stop after repeat-times. `--regtest` must be' |
| ' used with `--commission`.' |
| ) |
| _COMMISSION = flags.DEFINE_list( |
| name='commission', |
| default=None, |
| help=_COMMISSIONING_FLAG_USAGE_GUIDE, |
| ) |
| _DECOMMISSION = flags.DEFINE_string( |
| name='decommission', |
| default='', |
| help='Use --decommission {DeviceName} to remove the device on GHA.', |
| ) |
| _RUN_REGRESSION_TESTS = flags.DEFINE_boolean( |
| name='regtest', |
| default=False, |
| help=_REGRESSION_TESTS_FLAG_USAGE_GUIDE, |
| ) |
| _REPEAT = flags.DEFINE_integer( |
| name='repeat', |
| default=None, |
| help=_REGRESSION_TESTS_FLAG_USAGE_GUIDE, |
| ) |
| |
| |
| def _validate_commissioning_arg( |
| device_name: str, pairing_code: str, gha_room: str |
| ) -> None: |
| """Returns None if commissioning values are valid. |
| |
| Args: |
| device_name: Display name of commissioned device on GHA. |
| pairing_code: An 11-digit or 21-digit numeric code which contains the |
| information needed to commission a matter device. |
| gha_room: Assigned room of commissioned device on GHA. |
| |
| Raises: |
| ValueError: When any of passed arguments is invalid. |
| """ |
| device_name_match = re.fullmatch(r'\S{1,24}', device_name) |
| pairing_code_match = re.fullmatch(r'^(\d{11}|\d{21})$', pairing_code) |
| gha_room_match = re.fullmatch( |
| r'Attic|Back door|Backyard|Basement|Bathroom|Bedroom|Den|Dining' |
| r' Room|Entryway|Family Room|Front door|Front' |
| r' Yard|Garage|Hallway|Kitchen|Living Room|Master' |
| r' Bedroom|Office|Shed|Side door', |
| gha_room, |
| ) |
| |
| if not device_name_match: |
| raise ValueError( |
| 'Value of DeviceName is invalid. Device name should be no more than 24' |
| ' characters.' |
| ) |
| if not pairing_code_match: |
| raise ValueError( |
| 'Value of PairingCode is invalid. Paring code should be 11-digit or' |
| ' 21-digit numeric code.' |
| ) |
| if not gha_room_match: |
| raise ValueError( |
| 'Value of GHARoom is invalid. Valid values for GHARoom: Attic|Back' |
| ' door|Backyard|Basement|Bathroom|Bedroom|Den|Dining' |
| ' Room|Entryway|Family Room|Front door|Front' |
| ' Yard|Garage|Hallway|Kitchen|Living Room|Master' |
| ' Bedroom|Office|Shed|Side door' |
| ) |
| |
| |
| def get_android_device_ready(func: Callable[..., Any]) -> Callable[..., Any]: |
| """A decorator to check if an Android device can be controlled by installed apk. |
| |
| Args: |
| func: The function to be wrapped. |
| |
| Returns: |
| The wrapped function. |
| |
| Raises: |
| NoAndroidDeviceError: When no device is connected. |
| AndroidDeviceNotReadyError: When required snippet apk can not be loaded |
| on device. |
| """ |
| |
| @functools.wraps(func) |
| def wrapped_func(self, *args, **kwargs): |
| try: |
| self.load_device() |
| self.load_snippet() |
| except ( |
| errors.AdbError, |
| errors.AndroidDeviceNotReadyError, |
| errors.NoAndroidDeviceError, |
| ) as e: |
| raise errors.AndroidDeviceNotReadyError(f'{func.__name__} failed.') from e |
| |
| return func(self, *args, **kwargs) |
| |
| return wrapped_func |
| |
| |
| class UIAutomator: |
| """UI Automator controller in python layer.""" |
| |
| def __init__(self, logger: logging.Logger = logging.getLogger()): |
| """Inits UIAutomator. |
| |
| Interacts with android device by pre-installed apk. Can perform |
| methods provided by a snippet installed on the android device. |
| |
| Args: |
| logger: Injected logger, if None, specify the default(root) one. |
| """ |
| self._logger = logger |
| self._connected_device: android_device.AndroidDevice | None = None |
| |
| def load_device(self): |
| """Selects the first connected android device. |
| |
| Raises: |
| NoAndroidDeviceError: When no device is connected. |
| """ |
| try: |
| android_devices = android_device.get_all_instances() |
| except adb.AdbError as exc: |
| raise errors.AdbError( |
| 'Please install adb and add it to your PATH environment variable.' |
| ) from exc |
| |
| if not android_devices: |
| raise errors.NoAndroidDeviceError( |
| 'No Android device connected to the host computer.' |
| ) |
| self._connected_device = android_devices[0] |
| self._logger.info(f'connected device: [{self._connected_device}]') |
| |
| def load_snippet(self): |
| """Loads needed mobly snippet installed on android device. |
| |
| Raises: |
| NoAndroidDeviceError: When no device is connected. |
| AndroidDeviceNotReadyError: When required snippet apk can not be loaded |
| on device. |
| """ |
| if not self._connected_device: |
| raise errors.NoAndroidDeviceError( |
| 'No Android device connected to the host computer.' |
| ) |
| |
| is_apk_installed = ad.is_apk_installed( |
| self._connected_device, _MOBLY_SNIPPET_APK |
| ) |
| if not is_apk_installed or not ad.is_apk_version_correct( |
| self._connected_device, |
| _MOBLY_SNIPPET_APK, |
| version.VERSION, |
| ): |
| if is_apk_installed: |
| ad.uninstall_apk(self._connected_device, _MOBLY_SNIPPET_APK) |
| ad.install_apk(self._connected_device, self._get_mbs_apk_path()) |
| |
| try: |
| self._connected_device.load_snippet( |
| _MOBLY_SNIPPET_APK_NAME, _MOBLY_SNIPPET_APK |
| ) |
| except snippet_errors.ServerStartPreCheckError as exc: |
| # Raises when apk not be installed or instrumented on device. |
| raise errors.AndroidDeviceNotReadyError( |
| f"Check device({self._connected_device.device_info['serial']}) has" |
| ' installed required apk.' |
| ) from exc |
| except ( |
| snippet_errors.ServerStartError, |
| snippet_errors.ProtocolError, |
| ) as exc: |
| raise errors.AndroidDeviceNotReadyError( |
| f"Check device({self._connected_device.device_info['serial']}) can" |
| ' load required apk.' |
| ) from exc |
| except android_device.SnippetError as e: |
| # Error raises when registering a package twice or a package with |
| # duplicated name. Thus, It is okay to pass here as long as the snippet |
| # has been loaded. |
| self._logger.debug(str(e)) |
| |
| @get_android_device_ready |
| def commission_device( |
| self, |
| device_name: str, |
| pairing_code: str, |
| gha_room: str |
| ): |
| """Commissions a device through installed apk `mbs` on Google Home App. |
| |
| Args: |
| device_name: Display name of commissioned device on GHA. |
| pairing_code: An 11-digit or 21-digit numeric code which contains the |
| information needed to commission a matter device. |
| gha_room: Assigned room of commissioned device on GHA. |
| |
| Raises: |
| ValueError: When any of passed arguments is invalid. |
| MoblySnippetError: When running `commissionDevice` method in snippet apk |
| encountered an error. |
| """ |
| self._logger.info('Start commissioning the device.') |
| |
| _validate_commissioning_arg(device_name, pairing_code, gha_room) |
| |
| device_name_snake_case = f'_{inflection.underscore(device_name)}' |
| matter_device = { |
| 'id': device_name_snake_case, |
| 'name': device_name, |
| 'pairingCode': pairing_code, |
| 'roomName': gha_room, |
| } |
| |
| try: |
| self._connected_device.mbs.commissionDevice( |
| _GOOGLE_HOME_APP, matter_device |
| ) |
| self._logger.info('Successfully commission the device on GHA.') |
| # TODO(b/298903492): Narrow exception type. |
| except Exception as e: |
| raise errors.MoblySnippetError( |
| 'Unable to continue automated commissioning process on' |
| f' device({self._connected_device.device_info["serial"]}).' |
| ) from e |
| |
| @get_android_device_ready |
| def decommission_device(self, device_name: str): |
| """Removes given device through installed apk `mbs` on Google Home App. |
| |
| Args: |
| device_name: Display name of the device needs to be removed on GHA. |
| |
| Raises: |
| ValueError: When passed argument is invalid. |
| MoblySnippetError: When running `removeDevice` method in snippet apk |
| encountered an error. |
| """ |
| if not re.fullmatch(r'\S{1,24}', device_name): |
| raise ValueError( |
| 'Value of DeviceName is invalid. Device name should be no more than' |
| ' 24 characters.' |
| ) |
| |
| self._logger.info('Start removing the device.') |
| try: |
| self._connected_device.mbs.removeDevice(device_name) |
| self._logger.info('Successfully remove the device on GHA.') |
| # TODO(b/298903492): Narrow exception type. |
| except Exception as e: |
| raise errors.MoblySnippetError( |
| f'Unable to remove {device_name} from GHA' |
| f' on device({self._connected_device.device_info["serial"]}).' |
| ) from e |
| |
| def run_regression_tests( |
| self, repeat: int | None, *args: Any, |
| ) -> None: |
| """Executes automated regression tests. |
| |
| A single execution of both commissioning and decommissioning constitutes one |
| cycle. |
| |
| Args: |
| repeat: The value of flag `repeat`. If the value is None, regression |
| tests will be run repeatedly until keyboard interrupts. |
| *args: Any required value to run regression tests. |
| |
| Raises: |
| ValueError: When the value of flag `repeat` is not positive. |
| """ |
| if repeat and repeat <= 0: |
| raise ValueError('Number placed after `--repeat` must be positive.') |
| |
| failure_count = 0 |
| run_count = 0 |
| self._logger.info( |
| 'Start running regression tests' |
| f' {str(repeat) + " times" if repeat is not None else "continuously"}.' |
| ) |
| while repeat is None or run_count < repeat: |
| try: |
| device_name, _, _ = args |
| self.commission_device(*args) |
| self.decommission_device(device_name) |
| except errors.MoblySnippetError: |
| failure_count += 1 |
| except KeyboardInterrupt: |
| self._logger.info('Tests interrupted by keyboard.') |
| break |
| run_count += 1 |
| |
| self._logger.info( |
| 'Ran %d times. Passed %d times. Failed %d times.', |
| run_count, |
| run_count - failure_count, |
| failure_count, |
| ) |
| |
| def _get_mbs_apk_path(self) -> str: |
| return os.path.join( |
| os.path.dirname(os.path.abspath(__file__)), |
| 'android', |
| 'app', |
| 'snippet-0.2.0.apk', |
| ) |
| |
| |
| def _process_flags(ui_automator: UIAutomator) -> None: |
| """Does specific action based on given flag values.""" |
| if _COMMISSION.value: |
| if len(_COMMISSION.value) != 3: |
| raise flags.IllegalFlagValueError(_COMMISSIONING_FLAG_USAGE_GUIDE) |
| if _RUN_REGRESSION_TESTS.value: |
| ui_automator.run_regression_tests(_REPEAT.value, *_COMMISSION.value) |
| else: |
| ui_automator.commission_device(*_COMMISSION.value) |
| elif _DECOMMISSION.value: |
| ui_automator.decommission_device(_DECOMMISSION.value) |
| elif _RUN_REGRESSION_TESTS.value: |
| raise flags.IllegalFlagValueError(_REGRESSION_TESTS_FLAG_USAGE_GUIDE) |
| |
| |
| # TODO(b/309745485): Type of argv should be Sequence[str]. |
| def _main(argv) -> None: |
| if argv and len(argv) > 1: |
| raise app.UsageError(f'Too many command-line arguments: {argv!r}') |
| |
| ui_automator = UIAutomator() |
| _process_flags(ui_automator) |
| |
| |
| def run(): |
| app.run(_main) |
| |
| |
| if __name__ == '__main__': |
| run() |