diff --git a/openandroidinstaller/openandroidinstaller.py b/openandroidinstaller/openandroidinstaller.py index a6680738..fb62c760 100644 --- a/openandroidinstaller/openandroidinstaller.py +++ b/openandroidinstaller/openandroidinstaller.py @@ -193,21 +193,18 @@ def log_version_infos(bin_path): """Log the version infos of adb, fastboot and heimdall.""" # adb adbversion = [ - line for line in run_command("adb", ["version"], bin_path, enable_logging=False) + line for line in run_command("adb version", bin_path, enable_logging=False) ] logger.info(f"{adbversion[1].strip()}") # fastboot fbversion = [ line - for line in run_command( - "fastboot", ["--version"], bin_path, enable_logging=False - ) + for line in run_command("fastboot --version", bin_path, enable_logging=False) ] logger.info(f"{fbversion[1].strip()}") # heimdall hdversion = [ - line - for line in run_command("heimdall", ["info"], bin_path, enable_logging=False) + line for line in run_command("heimdall info", bin_path, enable_logging=False) ] logger.info(f"Heimdall version: {hdversion[1].strip()}") diff --git a/openandroidinstaller/tooling.py b/openandroidinstaller/tooling.py index 1c268b00..04199322 100644 --- a/openandroidinstaller/tooling.py +++ b/openandroidinstaller/tooling.py @@ -17,15 +17,14 @@ from pathlib import Path import subprocess from subprocess import ( - Popen, PIPE, STDOUT, CalledProcessError, - CompletedProcess, check_output, ) +import shlex from time import sleep -from typing import List, Optional +from typing import List, Optional, Union from loguru import logger @@ -33,10 +32,12 @@ def run_command( - tool: str, command: List[str], bin_path: Path, enable_logging: bool = True -) -> CompletedProcess: + command: str, bin_path: Path, enable_logging: bool = True +) -> Union[str, bool]: """Run a command with a tool (adb, fastboot, heimdall).""" - yield f"${' '.join([tool] + command )}" + yield f"${command}" + # split the command and extract the tool part + tool, *command = shlex.split(command) if tool not in ["adb", "fastboot", "heimdall"]: raise Exception(f"Unknown tool {tool}. Use adb, fastboot or heimdall.") if PLATFORM == "win32": @@ -50,7 +51,7 @@ def run_command( if enable_logging: logger.info(f"Run command: {full_command}") # run the command - with Popen( + with subprocess.Popen( full_command, stdout=PIPE, stderr=STDOUT, @@ -63,94 +64,105 @@ def run_command( logger.info(line.strip()) yield line.strip() + # finally return if the command was successful yield p.returncode == 0 +def add_logging(step_desc: str, return_if_fail: bool = False): + """Logging decorator to wrap functions that yield lines. + + Logs the `step_desc`. + """ + + def logging_decorator(func): + def logging(*args, **kwargs): + logger.info(f"{step_desc} - Paramters: {kwargs}") + for line in func(*args, **kwargs): + if (type(line) == bool) and not line: + logger.error(f"{step_desc} Failed!") + if return_if_fail: + yield False + return + yield line + + return logging + + return logging_decorator + + +@add_logging("Rebooting device with adb.") def adb_reboot(bin_path: Path) -> bool: """Run adb reboot on the device and return success.""" - logger.info("Rebooting device with adb.") - for line in run_command("adb", ["reboot"], bin_path): + for line in run_command("adb reboot", bin_path): yield line - if (type(line) == bool) and line: - logger.debug("Reboot failed.") - yield False - else: - yield True -def adb_reboot_bootloader(bin_path: Path) -> bool: +@add_logging("Rebooting device into bootloader with adb.", return_if_fail=True) +def adb_reboot_bootloader(bin_path: Path) -> Union[str, bool]: """Reboot the device into bootloader and return success.""" - logger.info("Rebooting device into bootloader with adb.") - for line in run_command("adb", ["reboot", "bootloader"], bin_path): - if (type(line) == bool) and not line: - logger.error("Reboot into bootloader failed.") + for line in run_command("adb reboot bootloader", bin_path): yield line sleep(1) -def adb_reboot_download(bin_path: Path) -> bool: +@add_logging("Rebooting device into download mode with adb.") +def adb_reboot_download(bin_path: Path) -> Union[str, bool]: """Reboot the device into download mode of samsung devices and return success.""" - logger.info("Rebooting device into download mode with adb.") - for line in run_command("adb", ["reboot", "download"], bin_path): + for line in run_command("adb reboot download", bin_path): yield line - if (type(line) == bool) and not line: - logger.error("Reboot into download mode failed.") - yield False - else: - # check if in download mode with heimdall? - yield True -def adb_sideload(bin_path: Path, target: str) -> bool: +@add_logging("Sideload the target to device with adb.") +def adb_sideload(bin_path: Path, target: str) -> Union[str, bool]: """Sideload the target to device and return success.""" - logger.info("Rebooting device into bootloader with adb.") - for line in run_command("adb", ["sideload", target], bin_path): + for line in run_command(f"adb sideload {target}", bin_path): + yield line + + +@add_logging("Activate sideloading in TWRP.", return_if_fail=True) +def activate_sideload(bin_path: Path) -> Union[str, bool]: + """Activate sideload with adb shell in twrp.""" + for line in run_command("adb shell twrp sideload", bin_path): yield line - if (type(line) == bool) and line: - logger.info(f"Sideloading {target} failed.") - yield False - else: - yield True def adb_twrp_copy_partitions(bin_path: Path, config_path: Path): # some devices like one plus 6t or motorola moto g7 power need the partitions copied to prevent a hard brick logger.info("Sideload copy_partitions script with adb.") # activate sideload - for line in run_command("adb", ["shell", "twrp", "sideload"], bin_path): + for line in activate_sideload(bin_path): yield line - if (type(line) == bool) and not line: - logger.error("Activating sideload failed.") - yield False - return # now sideload the script sleep(5) logger.info("Sideload the copy_partitions script") - for line in run_command( - "adb", - [ - "sideload", - f"{config_path.parent.joinpath(Path('copy-partitions-20220613-signed.zip'))}", - ], - bin_path, + for line in adb_sideload( + bin_path=bin_path, + target=f"{config_path.parent.joinpath(Path('copy-partitions-20220613-signed.zip'))}", ): yield line - if (type(line) == bool) and not line: - logger.error("Sideloading copy-partitions-20220613-signed.zip failed.") sleep(10) # reboot into the bootloader again - logger.info("Rebooting device into bootloader with adb.") - for line in run_command("adb", ["reboot", "bootloader"], bin_path): + for line in adb_reboot_bootloader(bin_path): yield line - if (type(line) == bool) and not line: - logger.error("Reboot into bootloader failed.") - yield False - return sleep(7) # Copy partitions end # return True +@add_logging("Perform a factory reset with adb and twrp.", return_if_fail=True) +def adb_twrp_format_data(bin_path: Path): + """Perform a factory reset with twrp and adb.""" + for line in run_command("adb shell twrp format data", bin_path): + yield line + + +@add_logging("Wipe the selected partition with adb and twrp.", return_if_fail=True) +def adb_twrp_wipe_partition(bin_path: Path, partition: str): + """Perform a factory reset with twrp and adb.""" + for line in run_command(f"adb shell twrp wipe {partition}", bin_path): + yield line + + def adb_twrp_wipe_and_install( bin_path: Path, target: str, @@ -166,94 +178,63 @@ def adb_twrp_wipe_and_install( logger.info("Wipe and format data with twrp, then install os image.") sleep(7) # now perform a factory reset - for line in run_command("adb", ["shell", "twrp", "format", "data"], bin_path): + for line in adb_twrp_format_data(bin_path): yield line - if (type(line) == bool) and not line: - logger.error("Formatting data failed.") - yield False - return + sleep(1) # wipe some partitions for partition in ["cache", "system"]: - for line in run_command("adb", ["shell", "twrp", "wipe", partition], bin_path): - yield not line + for line in adb_twrp_wipe_partition(bin_path=bin_path, partition=partition): + yield line sleep(1) - if (type(line) == bool) and not line: - logger.error(f"Wiping {partition} failed.") - yield False - return + # activate sideload logger.info("Wiping is done, now activate sideload.") - for line in run_command("adb", ["shell", "twrp", "sideload"], bin_path): + for line in activate_sideload(bin_path=bin_path): yield line - if (type(line) == bool) and not line: - logger.error("Activating sideload failed.") - yield False - return # now flash os image sleep(5) logger.info("Sideload and install os image.") - for line in run_command("adb", ["sideload", f"{target}"], bin_path): + for line in adb_sideload(bin_path=bin_path, target=target): yield line - if (type(line) == bool) and not line: - logger.error(f"Sideloading {target} failed.") - # TODO: this might sometimes think it failed, but actually it's fine. So skip for now. - # yield False - # return # wipe some cache partitions sleep(7) for partition in ["dalvik", "cache"]: - for line in run_command("adb", ["shell", "twrp", "wipe", partition], bin_path): + for line in run_command(f"adb shell twrp wipe {partition}", bin_path): yield line sleep(1) if (type(line) == bool) and not line: logger.error(f"Wiping {partition} failed.") # TODO: if this fails, a fix can be to just sideload something and then adb reboot for line in run_command( - "adb", - ["sideload", f"{config_path.parent.joinpath(Path('helper.txt'))}"], + f"adb sideload {config_path.parent.joinpath(Path('helper.txt'))}", bin_path, ): yield line - sleep(1) if (type(line) == bool) and not line: yield False break + sleep(2) # finally reboot into os or to fastboot for flashing addons sleep(7) if install_addons: if is_ab: # reboot into the bootloader again - logger.info("Rebooting device into bootloader with adb.") - for line in run_command("adb", ["reboot", "bootloader"], bin_path): + for line in adb_reboot_bootloader(bin_path): yield line - if (type(line) == bool) and not line: - logger.error("Reboot into bootloader failed.") - yield False - return sleep(3) # boot to TWRP again - logger.info("Boot custom recovery with fastboot.") - for line in run_command("fastboot", ["boot", f"{recovery}"], bin_path): + for line in fastboot_flash_recovery( + bin_path=bin_path, recovery=recovery, is_ab=is_ab + ): yield line - if (type(line) == bool) and not line: - logger.error("Reboot into bootloader failed.") - yield False - return sleep(7) else: # if not an a/b-device just stay in twrp pass else: - logger.info("Reboot into OS.") - for line in run_command("adb", ["reboot"], bin_path): + for line in adb_reboot(bin_path=bin_path): yield line - if (type(line) == bool) and not line: - logger.error("Rebooting failed.") - yield False - return - else: - yield True def adb_twrp_install_addons(bin_path: Path, addons: List[str], is_ab: bool) -> bool: @@ -267,143 +248,90 @@ def adb_twrp_install_addons(bin_path: Path, addons: List[str], is_ab: bool) -> b for addon in addons: # activate sideload logger.info("Activate sideload.") - for line in run_command("adb", ["shell", "twrp", "sideload"], bin_path): + for line in activate_sideload(bin_path=bin_path): yield line - if (type(line) == bool) and not line: - logger.error("Activating sideload failed.") - yield False - return sleep(5) # now flash os image - for line in run_command("adb", ["sideload", f"{addon}"], bin_path): + for line in adb_sideload(bin_path=bin_path, target=addon): yield line - if (type(line) == bool) and not line: - logger.error(f"Sideloading {addon} failed.") - # TODO: this might sometimes think it failed, but actually it's fine. So skip for now. - # yield False - # return sleep(7) # finally reboot into os if is_ab: # reboot into the bootloader again - logger.info("Rebooting device into bootloader with adb.") - for line in run_command("adb", ["reboot", "bootloader"], bin_path): + for line in adb_reboot_bootloader(bin_path=bin_path): yield line - if (type(line) == bool) and not line: - logger.error("Reboot into bootloader failed.") - yield False - return sleep(3) # switch active boot partition - logger.info("Switch active boot partition") - for line in run_command("fastboot", ["set_active", "other"], bin_path): + for line in fastboot_switch_partition(bin_path=bin_path): yield line - if (type(line) == bool) and not line: - logger.error("Switching boot partition failed.") - yield False - return sleep(1) - for line in run_command("fastboot", ["set_active", "other"], bin_path): + for line in fastboot_switch_partition(bin_path=bin_path): yield line - if (type(line) == bool) and not line: - logger.error("Switching boot partition failed.") - yield False - return sleep(1) # reboot with fastboot logger.info("Reboot into OS.") - for line in run_command("fastboot", ["reboot"], bin_path): + for line in fastboot_reboot(bin_path=bin_path): yield line - if (type(line) == bool) and not line: - logger.error("Rebooting failed.") - yield False - return - else: - yield True else: # reboot with adb - logger.info("Reboot into OS.") - for line in run_command("adb", ["reboot"], bin_path): + for line in adb_reboot(bin_path=bin_path): yield line - if (type(line) == bool) and not line: - logger.error("Rebooting failed.") - yield False - return - else: - yield True -def fastboot_unlock_with_code(bin_path: Path, unlock_code: str) -> bool: +@add_logging("Switch active boot partitions.", return_if_fail=True) +def fastboot_switch_partition(bin_path: Path) -> Union[str, bool]: + """Switch the active boot partition with fastboot.""" + for line in run_command("fastboot set_active other", bin_path): + yield line + + +@add_logging("Unlock the device with fastboot and code.") +def fastboot_unlock_with_code(bin_path: Path, unlock_code: str) -> Union[str, bool]: """Unlock the device with fastboot and code given.""" - logger.info(f"Unlock the device with fastboot and code: {unlock_code}.") - for line in run_command("fastboot", ["oem", "unlock", f"{unlock_code}"], bin_path): + for line in run_command(f"fastboot oem unlock {unlock_code}", bin_path): yield line - if (type(line) == bool) and not line: - logger.error(f"Unlocking with code {unlock_code} failed.") - yield False - else: - yield True -def fastboot_unlock(bin_path: Path) -> bool: +@add_logging("Unlock the device with fastboot without code.") +def fastboot_unlock(bin_path: Path) -> Union[str, bool]: """Unlock the device with fastboot and without code.""" - logger.info("Unlock the device with fastboot.") - for line in run_command("fastboot", ["flashing", "unlock"], bin_path): + for line in run_command("fastboot flashing unlock", bin_path): yield line - if (type(line) == bool) and not line: - logger.error("Unlocking failed.") - yield False - else: - yield True -def fastboot_oem_unlock(bin_path: Path) -> bool: +@add_logging("OEM unlocking the device with fastboot.") +def fastboot_oem_unlock(bin_path: Path) -> Union[str, bool]: """OEM unlock the device with fastboot and without code.""" - logger.info("OEM unlocking the device with fastboot.") - for line in run_command("fastboot", ["oem", "unlock"], bin_path): + for line in run_command("fastboot oem unlock", bin_path): yield line - if (type(line) == bool) and not line: - logger.error("OEM unlocking failed.") - yield False - else: - yield True -def fastboot_get_unlock_data(bin_path: Path) -> bool: +@add_logging("Get unlock data with fastboot") +def fastboot_get_unlock_data(bin_path: Path) -> Union[str, bool]: """Get the unlock data with fastboot""" - logger.info("Get unlock data with fastboot") - for line in run_command("fastboot", ["oem", "get_unlock_data"], bin_path): + for line in run_command("fastboot oem get_unlock_data", bin_path): yield line - if (type(line) == bool) and not line: - logger.error("Getting unlock data failed.") - yield False - else: - yield True -def fastboot_reboot(bin_path: Path) -> bool: +@add_logging("Rebooting device with fastboot.") +def fastboot_reboot(bin_path: Path) -> Union[str, bool]: """Reboot with fastboot""" - logger.info("Rebooting device with fastboot.") - for line in run_command("fastboot", ["reboot"], bin_path): + for line in run_command("fastboot reboot", bin_path): yield line - if (type(line) == bool) and not line: - logger.error("Rebooting with fastboot failed.") - yield False - else: - yield True -def fastboot_flash_recovery(bin_path: Path, recovery: str, is_ab: bool = True) -> bool: +@add_logging("Flash or boot custom recovery with fastboot.") +def fastboot_flash_recovery( + bin_path: Path, recovery: str, is_ab: bool = True +) -> Union[str, bool]: """Temporarily, flash custom recovery with fastboot.""" if is_ab: logger.info("Boot custom recovery with fastboot.") - for line in run_command("fastboot", ["boot", f"{recovery}"], bin_path): + for line in run_command(f"fastboot boot {recovery}", bin_path): yield line else: logger.info("Flash custom recovery with fastboot.") - for line in run_command( - "fastboot", ["flash", "recovery", f"{recovery}"], bin_path - ): + for line in run_command(f"fastboot flash recovery {recovery}", bin_path): yield line if (type(line) == bool) and not line: logger.error("Flashing recovery failed.") @@ -412,20 +340,14 @@ def fastboot_flash_recovery(bin_path: Path, recovery: str, is_ab: bool = True) - yield True # reboot logger.info("Boot into TWRP with fastboot.") - for line in run_command("fastboot", ["reboot", "recovery"], bin_path): + for line in run_command("fastboot reboot recovery", bin_path): yield line - if (type(line) == bool) and not line: - logger.error("Booting recovery failed.") - yield False - else: - yield True - def fastboot_flash_boot(bin_path: Path, recovery: str) -> bool: """Temporarily, flash custom recovery with fastboot to boot partition.""" logger.info("Flash custom recovery with fastboot.") - for line in run_command("fastboot", ["flash", "boot", f"{recovery}"], bin_path): + for line in run_command(f"fastboot flash boot {recovery}", bin_path): yield line if (type(line) == bool) and not line: logger.error("Flashing recovery failed.") @@ -434,7 +356,7 @@ def fastboot_flash_boot(bin_path: Path, recovery: str) -> bool: yield True # reboot logger.info("Boot into TWRP with fastboot.") - for line in run_command("fastboot", ["reboot"], bin_path): + for line in run_command("fastboot reboot", bin_path): yield line if (type(line) == bool) and not line: logger.error("Booting recovery failed.") @@ -443,18 +365,13 @@ def fastboot_flash_boot(bin_path: Path, recovery: str) -> bool: yield True -def heimdall_flash_recovery(bin_path: Path, recovery: str) -> bool: +@add_logging("Flash custom recovery with heimdall.") +def heimdall_flash_recovery(bin_path: Path, recovery: str) -> Union[str, bool]: """Temporarily, flash custom recovery with heimdall.""" - logger.info("Flash custom recovery with heimdall.") for line in run_command( - "heimdall", ["flash", "--no-reboot", "--RECOVERY", f"{recovery}"], bin_path + f"heimdall flash --no-reboot --RECOVERY {recovery}", bin_path ): yield line - if (type(line) == bool) and not line: - logger.error("Flashing recovery with heimdall failed.") - yield False - else: - yield True def search_device(platform: str, bin_path: Path) -> Optional[str]: diff --git a/openandroidinstaller/views/base.py b/openandroidinstaller/views/base.py index a45dcef8..f4975ba5 100644 --- a/openandroidinstaller/views/base.py +++ b/openandroidinstaller/views/base.py @@ -34,7 +34,7 @@ def __init__(self, state: AppState, image: str = "placeholder.png"): # right part of the display, add content here. self.right_view_header = Column(width=self.column_width, height=100, spacing=30) self.right_view = Column( - alignment="center", width=self.column_width, height=650 + alignment="center", width=self.column_width, height=650, scroll="auto" ) # left part of the display: used for displaying the images self.left_view = Column( diff --git a/openandroidinstaller/views/install_addons_view.py b/openandroidinstaller/views/install_addons_view.py index 54a7955b..3c8a15ea 100644 --- a/openandroidinstaller/views/install_addons_view.py +++ b/openandroidinstaller/views/install_addons_view.py @@ -63,6 +63,7 @@ def check_advanced_switch(e): logger.info("Disable advanced output.") self.state.advanced = False self.terminal_box.toggle_visibility() + self.right_view.update() self.advanced_switch = Switch( label="Advanced output", diff --git a/openandroidinstaller/views/install_view.py b/openandroidinstaller/views/install_view.py index dfd2d773..3f2ae4b8 100644 --- a/openandroidinstaller/views/install_view.py +++ b/openandroidinstaller/views/install_view.py @@ -63,6 +63,7 @@ def check_advanced_switch(e): logger.info("Disable advanced output.") self.state.advanced = False self.terminal_box.toggle_visibility() + self.right_view.update() self.advanced_switch = Switch( label="Advanced output", diff --git a/openandroidinstaller/views/step_view.py b/openandroidinstaller/views/step_view.py index db830c6c..e9557ef4 100644 --- a/openandroidinstaller/views/step_view.py +++ b/openandroidinstaller/views/step_view.py @@ -84,7 +84,6 @@ def build(self): # switch to enable advanced output - here it means show terminal input/output in tool def check_advanced_switch(e): - logger.info(self.advanced_switch.value) """Check the box to enable advanced output.""" if self.advanced_switch.value: logger.info("Enable advanced output.") @@ -196,6 +195,8 @@ def call_to_phone(self, e, command: str): Some parts of the command are changed by placeholders. """ + # clean the previous error display + self.error_text.value = "" # disable the call button while the command is running self.call_button.disabled = True # reset the progress indicators diff --git a/poetry.lock b/poetry.lock index 2bfe37ac..307a3078 100644 --- a/poetry.lock +++ b/poetry.lock @@ -554,6 +554,22 @@ pytest = ">=5.0" [package.extras] dev = ["pre-commit", "pytest-asyncio", "tox"] +[[package]] +name = "pytest-subprocess" +version = "1.5.0" +description = "A plugin to fake subprocess for pytest" +category = "main" +optional = false +python-versions = ">=3.6" + +[package.dependencies] +pytest = ">=4.0.0" + +[package.extras] +dev = ["changelogd", "nox"] +docs = ["changelogd", "furo", "sphinx", "sphinx-autodoc-typehints", "sphinxcontrib-napoleon"] +test = ["Pygments (>=2.0)", "anyio", "coverage", "docutils (>=0.12)", "pytest (>=4.0)", "pytest-asyncio (>=0.15.1)", "pytest-rerunfailures"] + [[package]] name = "pywin32-ctypes" version = "0.2.0" @@ -749,7 +765,7 @@ dev = ["black (>=19.3b0)", "pytest (>=4.6.2)"] [metadata] lock-version = "1.1" python-versions = "<3.11,>=3.9" -content-hash = "79178593143974480efa33e2fd27c48a5bb01438a09873a93b389496541b2834" +content-hash = "68d8e8e1fffc47b22e117cba8553b7a9577b91b706f752c43e317a45cc9a46bd" [metadata.files] altgraph = [ @@ -1570,6 +1586,10 @@ pytest-mock = [ {file = "pytest-mock-3.10.0.tar.gz", hash = "sha256:fbbdb085ef7c252a326fd8cdcac0aa3b1333d8811f131bdcc701002e1be7ed4f"}, {file = "pytest_mock-3.10.0-py3-none-any.whl", hash = "sha256:f4c973eeae0282963eb293eb173ce91b091a79c1334455acfac9ddee8a1c784b"}, ] +pytest-subprocess = [ + {file = "pytest-subprocess-1.5.0.tar.gz", hash = "sha256:d7693b96f588f39b84c7b2b5c04287459246dfae6be1dd4098937a728ad4fbe3"}, + {file = "pytest_subprocess-1.5.0-py3-none-any.whl", hash = "sha256:dfd75b10af6800a89a9b758f2e2eceff9de082a27bd1388521271b6e8bde298b"}, +] pywin32-ctypes = [ {file = "pywin32-ctypes-0.2.0.tar.gz", hash = "sha256:24ffc3b341d457d48e8922352130cf2644024a4ff09762a2261fd34c36ee5942"}, {file = "pywin32_ctypes-0.2.0-py2.py3-none-any.whl", hash = "sha256:9dc2d991b3479cc2df15930958b674a48a227d5361d413827a4cfd0b5876fc98"}, diff --git a/pyproject.toml b/pyproject.toml index a698a6bf..5b78472f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -24,6 +24,7 @@ py7zr = "^0.20.0" pytest-cov = "^4.0.0" pytest-mock = "^3.10.0" bandit = "^1.7.4" +pytest-subprocess = "^1.5.0" [tool.poetry.dev-dependencies] diff --git a/tests/conftest.py b/tests/conftest.py index 2bf0c8d9..f7640b61 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -11,7 +11,9 @@ # If not, see .""" # Author: Tobias Sterbak +import os import pytest +import subprocess from pathlib import Path diff --git a/tests/test_tooling.py b/tests/test_tooling.py index 0a3225ce..c2195d67 100644 --- a/tests/test_tooling.py +++ b/tests/test_tooling.py @@ -16,7 +16,42 @@ from pathlib import Path from subprocess import CalledProcessError -from openandroidinstaller.tooling import search_device +from openandroidinstaller.tooling import adb_reboot, search_device, check_ab_partition + + +def test_adb_reboot_success(fp): + """Test if rebooting with adb works fine.""" + + with fp.context() as nested_process: + nested_process.register( + ["test/path/to/tools/adb", "reboot"], stdout=bytes.fromhex("00") + ) + for line in adb_reboot(bin_path=Path("test/path/to/tools")): + print(line) + for_later = "error: no devices/emulators found" + assert line + + +def test_adb_reboot_failure(fp): + """Test if a fail in rebooting with adb is handled properly.""" + + def callback_function_with_kwargs(process, return_code): + process.returncode = return_code + + return_code = 1 + + with fp.context() as nested_process: + nested_process.register( + ["test/path/to/tools/adb", "reboot"], + stdout=[ + bytes("error: no devices/emulators found", encoding="utf-8"), + ], + callback=callback_function_with_kwargs, + callback_kwargs={"return_code": return_code}, + ) + for line in adb_reboot(bin_path=Path("test/path/to/tools")): + print(line) + assert not line def test_search_device_success(mocker): @@ -57,3 +92,43 @@ def patched_check_output(*args, **kwargs): ) assert device_code == None + + +def test_check_ab_device_is_ab(mocker): + """Test if checking for ab device works fine.""" + mocker.patch( + "openandroidinstaller.tooling.check_output", + return_value=b"[ro.boot.slot_suffix]: [_b]", + ) + + # test linux + is_ab = check_ab_partition( + platform="linux", bin_path=Path("openandroidinstaller/bin/") + ) + + assert is_ab + + # test windows + is_ab = check_ab_partition( + platform="windows", bin_path=Path("openandroidinstaller/bin/") + ) + + assert is_ab + + +def test_check_ab_device_not_ab(mocker): + """Test if checking for ab device returns False if it fails.""" + + def patched_check_output(*args, **kwargs): + raise CalledProcessError(returncode=1, cmd="output is None") + + mocker.patch( + "openandroidinstaller.tooling.check_output", + patched_check_output, + ) + + is_ab = check_ab_partition( + platform="linux", bin_path=Path("openandroidinstaller/bin/") + ) + + assert not is_ab