chore: major clean up in code base related to adb command execution (#307)

Also, fix screen recording which was broken

Author: ashishb

Makefile

@@ -1,6 +1,6 @@
 lint: lint_python lint_markdown
 
-test: test_python3
+test: test_python
 
 documentation:
 	pandoc --from=markdown --to=rst --output=docs/README.rst README.md && cd docs && make html
@@ -41,12 +41,9 @@ lint_python:
 	uv run ruff check --config pyproject.toml adbe
 	# E0602 is due to undefined variable unicode which is defined only for Python 2
 	# W0511 is fixme due to TODOs in the code.
-	# adbe/adbe.py:756:8: W0601: Global variable 'screen_record_file_path_on_device' undefined at the module level (global-variable-undefined)
-	# adbe/adbe.py:764:8: W0601: Global variable 'screen_record_file_path_on_device' undefined at the module level (global-variable-undefined)
-	# adbe/adbe.py:752:4: W0621: Redefining name 'screen_record_file_path_on_device' from outer scope (line 759) (redefined-outer-name)
 	# C0111: Missing function docstring (missing-docstring)
 	uv run -- pylint --disable=C0103,C0111,C0209,W1514 release.py
-	uv run -- pylint adbe/*.py tests/*.py --disable=R0123,R0911,R0912,R0914,R0915,R1705,R1710,C0103,C0111,C0209,C0301,C0302,C1801,W0511,W0621,W0601,W0602,W0603
+	uv run -- pylint adbe/*.py tests/*.py --disable=R0123,R0911,R0912,R0914,R0915,R1705,R1710,C0103,C0111,C0209,C0301,C0302,C1801,W0511,W0602,W0603
 	uv run -- flake8 adbe --count --ignore=F401,E126,E501,W503 --show-source --statistics
 	# Default complexity limit is 10
 	# Default line length limit is 127

adbe/adb_enhanced.py

@@ -27,6 +27,7 @@
         execute_adb_command2,
         execute_adb_shell_command,
         execute_adb_shell_command2,
+        execute_adb_shell_command3,
         execute_file_related_adb_shell_command,
         get_adb_shell_property,
         get_device_android_api_version,
@@ -49,6 +50,7 @@
         execute_adb_command2,
         execute_adb_shell_command,
         execute_adb_shell_command2,
+        execute_adb_shell_command3,
         execute_file_related_adb_shell_command,
         get_adb_shell_property,
         get_device_android_api_version,
@@ -432,13 +434,12 @@ def handle_list_devices() -> None:
 
 def _get_device_serials() -> list[str]:
     cmd = "devices -l"
-    return_code, stdout, stderr = execute_adb_command2(cmd)
-    if return_code != 0:
-        print_error_and_exit(f"Failed to execute command {cmd}, error: {stderr} ")
+    result = execute_adb_command2(cmd)
+    assert result.return_code == 0, f"Failed to execute command {cmd}, error: {result.stderr}"
 
     device_serials = []
     # Skip the first line, it says "List of devices attached"
-    device_infos = stdout.split("\n")[1:]
+    device_infos = result.stdout.split("\n")[1:]
 
     if len(device_infos) == 0 or (
             len(device_infos) == 1 and len(device_infos[0]) == 0):
@@ -522,18 +523,16 @@ def dump_ui(xml_file: str) -> None:
     cmd3 = f"rm {tmp_file}"
 
     print_verbose(f"Writing UI to {tmp_file}")
-    return_code, _, stderr = execute_adb_shell_command2(cmd1)
-    if return_code != 0:
-        print_error_and_exit(f'Failed to execute "{cmd1}", stderr: "{stderr}"')
+    result = execute_adb_shell_command3(cmd1)
+    assert result.return_code == 0, f"Failed to execute command {cmd1}, error: {result.stderr}"
 
     print_verbose(f"Pulling file {xml_file}")
-    return_code, _, stderr = execute_adb_command2(cmd2)
+    result = execute_adb_command2(cmd2)
+    assert result.return_code == 0, f"Failed to execute command {cmd2}, error: {result.stderr}"
     print_verbose(f"Deleting file {tmp_file}")
-    execute_adb_shell_command2(cmd3)
-    if return_code != 0:
-        print_error_and_exit(f"Failed to fetch file {tmp_file}")
-    else:
-        print_message(f'XML UI dumped to {xml_file}, you might want to format it using "xmllint --format {xml_file}"')
+    result = execute_adb_shell_command3(cmd3)
+    assert result.return_code == 0, f"Failed to execute command {cmd3}, error: {result.stderr}"
+    print_message(f'XML UI dumped to {xml_file}, you might want to format it using "xmllint --format {xml_file}"')
 
 
 @ensure_package_exists
@@ -611,14 +610,14 @@ def force_rtl(*, turn_on: bool) -> None:
 def dump_screenshot(filepath: str) -> None:
     screenshot_file_path_on_device = _create_tmp_file("screenshot", "png")
     dump_cmd = f"screencap -p {screenshot_file_path_on_device} "
-    return_code, stdout, stderr = execute_adb_shell_command2(dump_cmd)
-    if return_code != 0:
-        print_error_and_exit(
-            f"Failed to capture the screenshot: (stdout: {stdout}, stderr: {stderr})")
+    result = execute_adb_shell_command3(dump_cmd)
+    assert result.return_code == 0, f"Failed to capture screenshot, error: {result.stderr}"
     pull_cmd = f"pull {screenshot_file_path_on_device} {filepath}"
-    execute_adb_command2(pull_cmd)
+    result = execute_adb_command2(pull_cmd)
+    assert result.return_code == 0, f"Failed to pull screenshot file, error: {result.stderr}"
     del_cmd = f"rm {screenshot_file_path_on_device}"
-    execute_adb_shell_command2(del_cmd)
+    result = execute_adb_shell_command3(del_cmd)
+    assert result.return_code == 0, f"Failed to delete screenshot file from device, error: {result.stderr}"
 
 
 def dump_screenrecord(filepath: str) -> None:
@@ -632,12 +631,10 @@ def dump_screenrecord(filepath: str) -> None:
 
     original_sigint_handler = None
 
-    def _start_recording() -> str:
+    def _start_recording(screen_record_file_path: str) -> None:
         print_message("Recording video, press Ctrl+C to end...")
-        tmp_file_path = _create_tmp_file("screenrecord", "mp4")
-        dump_cmd = f"screenrecord --verbose {tmp_file_path} "
+        dump_cmd = f"screenrecord --verbose {screen_record_file_path} "
         execute_adb_shell_command2(dump_cmd)
-        return tmp_file_path
 
     def _pull_and_delete_file_from_device(screen_record_file_path: str) -> None:
         print_message(f"Saving recording to {filepath}")
@@ -674,7 +671,8 @@ def signal_handler(_sig: int, _frame: Any) -> None:
     original_sigint_handler = signal.getsignal(signal.SIGINT)
     signal.signal(signal.SIGINT, signal_handler)
 
-    screen_record_file_path_on_device = _start_recording()
+    screen_record_file_path_on_device = _create_tmp_file("screenrecord", "mp4")
+    _start_recording(screen_record_file_path_on_device)
 
 
 def get_mobile_data_saver_state() -> str:
@@ -1388,9 +1386,9 @@ def push_file(local_file_path: str, remote_file_path: str) -> None:
     cp_cmd = f"cp {tmp_file} {remote_file_path}"
     rm_cmd = f"rm {tmp_file}"
 
-    return_code, _, stderr = execute_adb_command2(push_cmd)
-    if return_code != 0:
-        print_error_and_exit(f"Failed to push file, error: {stderr}")
+    result = execute_adb_command2(push_cmd)
+    if result.return_code != 0:
+        print_error_and_exit(f"Failed to push file, error: {result.stderr}")
         return
 
     execute_file_related_adb_shell_command(cp_cmd, remote_file_path)
@@ -1590,9 +1588,9 @@ def print_app_signature(app_name: str) -> None:
     with tempfile.NamedTemporaryFile(prefix=app_name, suffix=".apk") as tmp_apk_file:
         tmp_apk_file_name = tmp_apk_file.name
         adb_cmd = f"pull {apk_path} {tmp_apk_file_name}"
-        return_code, _, stderr = execute_adb_command2(adb_cmd)
-        if return_code != 0:
-            print_error_and_exit(f"Failed to pull file {apk_path}, stderr: {stderr}")
+        result = execute_adb_command2(adb_cmd)
+        if result.return_code != 0:
+            print_error_and_exit(f"Failed to pull file {apk_path}, stderr: {result.stderr}")
             return
 
         dir_of_this_script = os.path.split(__file__)[0]
@@ -1669,9 +1667,9 @@ def backup_func() -> None:
 def perform_install(file_path: str) -> None:
     print_verbose(f"Installing {file_path}")
     # -r: replace existing application
-    return_code, _, stderr = execute_adb_command2(f"install -r {file_path}")
-    if return_code != 0:
-        print_error(f"Failed to install {file_path}, stderr: {stderr}")
+    result = execute_adb_command2(f"install -r {file_path}")
+    if result.return_code != 0:
+        print_error(f"Failed to install {file_path}, stderr: {result.stderr}")
 
 
 @ensure_package_exists
@@ -1697,23 +1695,9 @@ def perform_uninstall(app_name: str, first_user: bool) -> None:
         print_error(f"Failed to uninstall {app_name}, stderr: {stderr}")
 
 
-def _get_window_size() -> tuple[int, int]:
-    adb_cmd = "shell wm size"
-    _, result, _ = execute_adb_command2(adb_cmd)
-
-    if result is None:
-        return -1, -1
-
-    regex_data = re.search(r"size: ([0-9]+)x([0-9]+)", result)
-    if regex_data is None:
-        return -1, -1
-
-    return int(regex_data.group(1)), int(regex_data.group(2))
-
-
 def _perform_tap(x: int, y: int) -> None:
     adb_shell_cmd = f"input tap {x:d} {y:d}"
-    execute_adb_shell_command2(adb_shell_cmd)
+    execute_adb_shell_command3(adb_shell_cmd)
 
 
 # Deprecated
@@ -1771,14 +1755,14 @@ def enable_wireless_debug() -> bool:
 
     ip = matching[0]
 
-    code, _, stderr = execute_adb_command2("tcpip 5555")
-    if code != 0:
+    result = execute_adb_command2("tcpip 5555")
+    if result.return_code != 0:
         print_error_and_exit(f"Failed to switch device {ip} to wireless debug mode, "
-                             f"stderr: {stderr}")
+                             f"stderr: {result.stderr}")
 
-    code, _, stderr = execute_adb_command2(f"connect {ip}")
-    if code != 0:
-        print_error_and_exit(f"Cannot enable wireless debugging. Error: {stderr}")
+    result = execute_adb_command2(f"connect {ip}")
+    if result.return_code != 0:
+        print_error_and_exit(f"Cannot enable wireless debugging. Error: {result.stderr}")
         return False
     print_message(f"Connected via IP now you can disconnect the cable\nIP: {ip}")
     return True
@@ -1805,9 +1789,9 @@ def disable_wireless_debug() -> None:
     result = True
 
     for ip in ip_list:
-        code, _, stderr = execute_adb_command2(f"disconnect {ip}")
-        if code != 0:
-            print_error(f"Failed to disconnect {ip}: {stderr}")
+        result = execute_adb_command2(f"disconnect {ip}")
+        if result.return_code != 0:
+            print_error(f"Failed to disconnect {ip}: {result.stderr}")
             result = False
         else:
             print_message(f"Disconnected {ip}")
@@ -2090,10 +2074,10 @@ def print_state_change_info(state_name: str, old_state: str | int | bool, new_st
 
 def print_display_size() -> None:
     # Ref: https://developer.android.com/training/multiscreen/screendensities
-    cmd = "shell wm density"
-    return_code, stdout, stderr = execute_adb_command2(cmd)
-    assert return_code == 0, f"Failed to get display size, stderr: {stderr}"
-    display_size = re.search(r"Physical density: (\d+)", stdout)
+    cmd = "wm density"
+    result = execute_adb_shell_command3(cmd)
+    assert result.return_code == 0, f"Failed to get display size, stderr: {result.stderr}"
+    display_size = re.search(r"Physical density: (\d+)", result.stdout)
     if display_size is None:
         print_error_and_exit("Failed to get display size")
 

adbe/adb_helper.py

@@ -11,6 +11,13 @@
     from output_helper import print_error, print_error_and_exit, print_verbose
 
 
+@dataclasses.dataclass
+class CommandResult:
+    return_code: int
+    stdout: str | None
+    stderr: str | None
+
+
 @dataclasses.dataclass
 class _Settings:
     adb_prefix: str = "adb"
@@ -36,20 +43,33 @@ def set_adb_prefix(adb_prefix: str) -> None:
 
 
 def get_adb_shell_property(property_name: str, device_serial: str | None = None) -> str | None:
-    _, stdout, _ = execute_adb_shell_command2(f"getprop {property_name}", device_serial=device_serial)
-    return stdout
+    result = execute_adb_shell_command3(f"getprop {property_name}", device_serial=device_serial)
+    if result.return_code != 0:
+        print_error(f"Unable to get property {property_name} from device, return code: {result.return_code}")
+        return None
 
+    return result.stdout
 
-def execute_adb_shell_command2(
+
+def execute_adb_shell_command3(
         adb_cmd: str, piped_into_cmd: bool | None = None, ignore_stderr: bool = False,
-        device_serial: str | None = None) -> tuple[int, str | None, str]:
+        device_serial: str | None = None) -> CommandResult:
     return execute_adb_command2(f"shell {adb_cmd}", piped_into_cmd=piped_into_cmd,
                                 ignore_stderr=ignore_stderr, device_serial=device_serial)
 
 
-def execute_adb_command2(
+# Deprecated function, use execute_adb_shell_command3 instead
+def execute_adb_shell_command2(
         adb_cmd: str, piped_into_cmd: bool | None = None, ignore_stderr: bool = False,
         device_serial: str | None = None) -> tuple[int, str | None, str]:
+    result = execute_adb_shell_command3(
+        adb_cmd, piped_into_cmd=piped_into_cmd, ignore_stderr=ignore_stderr, device_serial=device_serial)
+    return result.return_code, result.stdout, result.stderr
+
+
+def execute_adb_command2(
+        adb_cmd: str, piped_into_cmd: bool | None = None, ignore_stderr: bool = False,
+        device_serial: str | None = None) -> CommandResult:
     """
     :param adb_cmd: command to run inside the adb shell (so, don't prefix it with "adb")
     :param piped_into_cmd: command to pipe the output of this command into
@@ -69,10 +89,12 @@ def execute_adb_command2(
     with subprocess.Popen(final_cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE) as ps1:
         stdout_data, stderr_data = ps1.communicate()
         return_code = ps1.returncode
+
     try:
         stdout_data = stdout_data.decode("utf-8")
     except UnicodeDecodeError:
         print_error("Unable to decode data as UTF-8, defaulting to printing the binary data")
+
     stderr_data = stderr_data.decode("utf-8")
 
     _check_for_adb_not_found_error(stderr_data)
@@ -82,13 +104,13 @@ def execute_adb_command2(
         print_error(stderr_data)
 
     if not stdout_data:
-        return return_code, None, stderr_data
+        return CommandResult(return_code=return_code, stdout=None, stderr=stderr_data)
 
     # stdout_data is not None
     if isinstance(stdout_data, bytes):
         print_verbose(f'Result is "{stdout_data}"')
-        return return_code, stdout_data, stderr_data
-    # str for Python 3, this used to be unicode type for python 2
+        return CommandResult(return_code=return_code, stdout=stdout_data, stderr=stderr_data)
+
     if isinstance(stdout_data, str):
         output = ""
         first_line = True
@@ -104,16 +126,18 @@ def execute_adb_command2(
             else:
                 output += "\n" + line
         print_verbose(f'Result is "{output}"')
-        return return_code, output, stderr_data
+        return CommandResult(return_code=return_code, stdout=output, stderr=stderr_data)
+
     print_error_and_exit(f"stdout_data is weird type: {type(stdout_data)}")
-    return None
+    return CommandResult(return_code=return_code, stdout=None, stderr="")
 
 
+# Deprecated function, use execute_adb_shell_command3 instead
 def execute_adb_shell_command(adb_cmd: str, piped_into_cmd: str | None = None, ignore_stderr: bool = False,
                               device_serial: str | None = None) -> str:
-    _, stdout, _ = execute_adb_command2(
+    result = execute_adb_command2(
         f"shell {adb_cmd}", piped_into_cmd, ignore_stderr, device_serial=device_serial)
-    return stdout
+    return result.stdout
 
 
 def execute_file_related_adb_shell_command(
@@ -137,19 +161,21 @@ def execute_file_related_adb_shell_command(
         print_verbose(f'Attempt {attempt_count}/{len(adb_cmds_prefix)}: "{adb_cmd_prefix}"')
         attempt_count += 1
         adb_cmd = f"{adb_cmd_prefix} {adb_shell_cmd}"
-        return_code, stdout, stderr = execute_adb_command2(
+        result = execute_adb_command2(
             adb_cmd, piped_into_cmd, ignore_stderr, device_serial=device_serial)
 
-        if stderr.find(file_not_found_message) >= 0:
+        if result.stderr.find(file_not_found_message) >= 0:
             print_error(f"File not found: {file_path}")
-            return stderr
-        if stderr.find(is_a_directory_message) >= 0:
+            return result.stderr
+        if result.stderr.find(is_a_directory_message) >= 0:
             print_error(f"{file_path} is a directory")
-            return stderr
+            return result.stderr
 
         api_version = get_device_android_api_version()
-        if api_version >= _MIN_VERSION_ABOVE_WHICH_ADB_SHELL_RETURNS_CORRECT_EXIT_CODE and return_code == 0:
-            return stdout
+        if api_version >= _MIN_VERSION_ABOVE_WHICH_ADB_SHELL_RETURNS_CORRECT_EXIT_CODE and result.return_code == 0:
+            return result.stdout
+
+        stdout = result.stdout
 
     return stdout