Skip to content

Commit

Permalink
style adjusted
Browse files Browse the repository at this point in the history
  • Loading branch information
Abdullah-Albanna committed Oct 29, 2024
1 parent cb9f1e0 commit f0fa896
Showing 1 changed file with 17 additions and 11 deletions.
28 changes: 17 additions & 11 deletions pymobiledevice3/services/installation_proxy.py
Original file line number Diff line number Diff line change
@@ -1,9 +1,9 @@
import os
from io import BytesIO
from pathlib import Path
from tempfile import TemporaryDirectory
from typing import Callable, Optional, Literal, Union
from zipfile import ZIP_DEFLATED, ZipFile, BadZipFile
from io import BytesIO
from typing import Callable, Literal, Optional, Union
from zipfile import ZIP_DEFLATED, BadZipFile, ZipFile

from parameter_decorators import str_to_path

Expand Down Expand Up @@ -33,6 +33,7 @@ def create_ipa_contents_from_directory(directory: str) -> bytes:
arcname=f'{payload_prefix}/{os.path.relpath(full_path, directory)}')
return zip_path.read_bytes()


def classify_zip_file(zip_bytes: bytes) -> Literal['ipa', 'ipcc']:
""" checks the zipped bytes if it's a .ipcc or .ipa """
try:
Expand All @@ -52,6 +53,7 @@ def classify_zip_file(zip_bytes: bytes) -> Literal['ipa', 'ipcc']:
except BadZipFile:
raise AppInstallError('invalid bytes package')


class InstallationProxyService(LockdownService):
SERVICE_NAME = 'com.apple.mobile.installation_proxy'
RSD_SERVICE_NAME = 'com.apple.mobile.installation_proxy.shim.remote'
Expand Down Expand Up @@ -116,7 +118,7 @@ def uninstall(self, bundle_identifier: str, options: Optional[dict] = None, hand
self.send_cmd_for_bundle_identifier(bundle_identifier, 'Uninstall', options, handler, args)

def install_package(self, package: Union[str, bytes], cmd: str = 'Install', options: Optional[dict] = None,
handler: Callable = None, *args) -> None:
handler: Callable = None, *args) -> None:
""" upload given ipa/ipcc onto device and install it """
if options is None:
options = {}
Expand All @@ -125,7 +127,7 @@ def install_package(self, package: Union[str, bytes], cmd: str = 'Install', opti
self.install_from_local(package, cmd, options, handler, *args)
elif isinstance(package, bytes):
self.install_from_bytes(package, cmd, options, handler, *args)

def install_from_bytes(self, package_bytes: bytes, cmd: str, options: Optional[dict], handler: Callable, *args) -> None:
ipcc_mode = classify_zip_file(package_bytes) == 'ipcc'

Expand All @@ -139,7 +141,7 @@ def install_from_bytes(self, package_bytes: bytes, cmd: str, options: Optional[d
else:
self.upload_ipcc_as_folder(package_bytes, afc)

self.send_package(cmd, options, handler, ipcc_mode *args)
self.send_package(cmd, options, handler, ipcc_mode, *args)

@str_to_path('package_path')
def install_from_local(self, package_path: Path, cmd: str, options: Optional[dict], handler: Callable, *args) -> None:
Expand All @@ -163,13 +165,17 @@ def install_from_local(self, package_path: Path, cmd: str, options: Optional[dic
else:
self.upload_ipcc_as_folder(package_path, afc)

self.send_package(cmd, options, handler, ipcc_mode *args)
self.send_package(cmd, options, handler, ipcc_mode, *args)

def send_package(self, cmd: str, options: Optional[dict], handler: Callable, ipcc_mode: bool = False, *args):
self.service.send_plist({'Command': cmd,
'ClientOptions': options,
'PackagePath': TEMP_REMOTE_IPCC_FOLDER if ipcc_mode
else TEMP_REMOTE_IPA_FILE})
self.service.send_plist({
'Command': cmd,
'ClientOptions': options,
'PackagePath': (
TEMP_REMOTE_IPCC_FOLDER if ipcc_mode
else TEMP_REMOTE_IPA_FILE
)
})

self._watch_completion(handler, ipcc_mode, args)

Expand Down

0 comments on commit f0fa896

Please sign in to comment.