From e1772fa9a2b3adc897f8848978b9697c124a675e Mon Sep 17 00:00:00 2001 From: Antonio Paolillo Date: Mon, 30 Sep 2024 20:42:51 +0200 Subject: [PATCH] hdc: reformat files & small fixes (#127) Signed-off-by: Antonio Paolillo --- benchkit/hdc/__init__.py | 92 ++++++++++++++++++---------------------- tests/test_hdc_api.py | 70 +++++++++++++++--------------- 2 files changed, 77 insertions(+), 85 deletions(-) diff --git a/benchkit/hdc/__init__.py b/benchkit/hdc/__init__.py index 6744acc..c33940a 100644 --- a/benchkit/hdc/__init__.py +++ b/benchkit/hdc/__init__.py @@ -3,22 +3,25 @@ Module to handle hdc (OpenHarmony Device Connector) interactions between host and remote phone. HDC documentation: https://docs.openharmony.cn/pages/v5.0/en/application-dev/dfx/hdc.md """ -from typing import Iterable, Optional, Callable, List from enum import Enum +from platform import system as os_system +from typing import Callable, Iterable, List, Optional +from benchkit.dependencies.executables import ExecutableDependency +from benchkit.dependencies.packages import Dependency from benchkit.shell.shell import get_args, shell_out from benchkit.utils.types import Command, PathType -from benchkit.dependencies.packages import Dependency -from benchkit.dependencies.executables import ExecutableDependency class HDCError(Exception): """Handle for errors from hdc.""" + class DeviceIdentifierKind(Enum): - IP_AND_PORT = 0, - SERIAL = 1, - DONT_CARE = 2, + IP_AND_PORT = 0 + SERIAL = 1 + DONT_CARE = 2 + class HDCDevice: """Representation of a device connected through hdc.""" @@ -33,7 +36,7 @@ def __init__( def __str__(self) -> str: return f"{self.identifier}" - + class OpenHarmonyDeviceConnector: """Operations with the phone for high-level hdc operations.""" @@ -51,48 +54,43 @@ def __init__( self._keep_connected = keep_connected self._wait_connected = wait_connected self._expected_os = expected_os + self._bin = "hdc.exe" if "Windows" == os_system() else "hdc" - - def __init__( - self, + @staticmethod + def from_device( device: HDCDevice, keep_connected: bool = False, wait_connected: bool = False, expected_os: Optional[str] = None, - ) -> None: - self.identifier = device.identifier - self.kind = device.kind - self._keep_connected = keep_connected - self._wait_connected = wait_connected - self._expected_os = expected_os - - - @staticmethod - def dependencies() -> List[Dependency]: - return [ - ExecutableDependency("hdc.exe") - ] + ) -> "OpenHarmonyDeviceConnector": + return OpenHarmonyDeviceConnector( + identifier=device.identifier, + kind=device.kind, + keep_connected=keep_connected, + wait_connected=wait_connected, + expected_os=expected_os, + ) + def dependencies(self) -> List[Dependency]: + return [ExecutableDependency(self._bin)] def _find_device(self) -> Optional[HDCDevice]: devices = [dev for dev in self._devices() if dev.identifier == self.identifier] match len(devices): case 0: return None - case 1: + case 1: return devices[0] case _: raise ValueError("Wrong device list.") - - @staticmethod - def _devices() -> Iterable[str]: + def _devices(self) -> List[HDCDevice]: """Get list of devices recognized by hdc. Returns: Iterable[HDCDevice]: list of devices recognized by hdc. """ - output = OpenHarmonyDeviceConnector._host_shell_out(command="hdc.exe list targets") + output = OpenHarmonyDeviceConnector._host_shell_out(command=f"{self._bin} list targets") device_ids = output.strip().splitlines() devices = [] for dev in device_ids: @@ -100,22 +98,22 @@ def _devices() -> Iterable[str]: devices.append(HDCDevice(dev, DeviceIdentifierKind.IP_AND_PORT)) else: devices.append(HDCDevice(dev, DeviceIdentifierKind.SERIAL)) - + return devices - - - @staticmethod - def query_devices(filter: Callable[[HDCDevice], bool] = lambda _: True) -> Iterable[HDCDevice]: + + def query_devices( + self, + filter_callback: Callable[[HDCDevice], bool] = lambda _: True, + ) -> Iterable[HDCDevice]: """Get filtered list of devices recognized by hdc. Returns: Iterable[HDCDevice]: filtered list of devices recognized by hdc. """ - devices = OpenHarmonyDeviceConnector._devices() - filtered = [dev for dev in devices if filter(dev)] + devices = self._devices() + filtered = [dev for dev in devices if filter_callback(dev)] return filtered - @staticmethod def _host_shell_out( command: Command, @@ -130,8 +128,7 @@ def _host_shell_out( print_output=print_output, ) return output - - + def _target_shell_out( self, command: Command, @@ -141,12 +138,7 @@ def _target_shell_out( dir_args = ["cd", f"{current_dir}", "&&"] if current_dir is not None else [] command_args = dir_args + get_args(command) - hdc_command = [ - "hdc.exe", - "-t", - f"{self.identifier}", - "shell" - ] + command_args + hdc_command = [f"{self._bin}", "-t", f"{self.identifier}", "shell"] + command_args output = shell_out( command=hdc_command, @@ -156,7 +148,6 @@ def _target_shell_out( return output - def shell_out( self, command: Command, @@ -180,7 +171,6 @@ def shell_out( current_dir=current_dir, output_is_log=output_is_log, ) - def push( self, @@ -201,7 +191,7 @@ def push( "file", "send", f"{local_path}", - f"{remote_path}" + f"{remote_path}", ] self._host_shell_out(command=command) @@ -217,13 +207,13 @@ def pull( remote_path (PathType): path on the device where the file is. local_path (PathType): path where to pull the file on the host. """ - commmand = [ - "hdc.exe", + command = [ + f"{self._bin}", "-t", f"{self.identifier}", "file", "recv", f"{remote_path}", - f"{local_path}" + f"{local_path}", ] - self._host_shell_out(command=commmand) + self._host_shell_out(command=command) diff --git a/tests/test_hdc_api.py b/tests/test_hdc_api.py index 8590526..6d2071c 100644 --- a/tests/test_hdc_api.py +++ b/tests/test_hdc_api.py @@ -1,34 +1,36 @@ -#!/usr/bin/env python3 -# SPDX-License-Identifier: MIT -""" -Module to test the HDC api basics -""" - -import os -from benchkit.hdc import * - -HOST_FILE = ".gitignore" -TMP_FILE = "cloned-gitignore" -TARGET_FILE = "/storage/media/100/local/testing/.gitignore" - -def main() -> None: - device = OpenHarmonyDeviceConnector.query_devices(lambda _: True)[0] - print(device) - - bridge = OpenHarmonyDeviceConnector(device) - output = bridge._target_shell_out("cd data && ls") - print(output) - - bridge.push(HOST_FILE, TARGET_FILE) - bridge.pull(TARGET_FILE, TMP_FILE) - - f = open(TMP_FILE, 'r') - content=f.read() - print(content) - f.close() - - os.remove(f.name) - - -if __name__ == '__main__': - main() +#!/usr/bin/env python3 +# SPDX-License-Identifier: MIT +""" +Module to test the HDC api basics +""" + +import os + +from benchkit.hdc import * + +HOST_FILE = ".gitignore" +TMP_FILE = "cloned-gitignore" +TARGET_FILE = "/storage/media/100/local/testing/.gitignore" + + +def main() -> None: + device = OpenHarmonyDeviceConnector.query_devices(lambda _: True)[0] + print(device) + + bridge = OpenHarmonyDeviceConnector(device) + output = bridge._target_shell_out("cd data && ls") + print(output) + + bridge.push(HOST_FILE, TARGET_FILE) + bridge.pull(TARGET_FILE, TMP_FILE) + + f = open(TMP_FILE, "r") + content = f.read() + print(content) + f.close() + + os.remove(f.name) + + +if __name__ == "__main__": + main()