kam-mobile/tools/device.py

"""Install and launch the current Android application package."""
from __future__ import annotations
import argparse, json, shutil, subprocess, xml.etree.ElementTree as ET
from pathlib import Path
from common import (ROOT, CPUS, ProbeError, sha256, CURRENT_BUILD_METADATA,
                    CURRENT_PACKAGE_METADATA, CURRENT_APK, CURRENT_BUILD_SCHEMA,
                    CURRENT_PACKAGE_SCHEMA)

class Adb:
    def __init__(self, serial: str | None): self.base=['adb']+(['-s',serial] if serial else [])
    def run(self,*args:str,check:bool=True,timeout:int=120)->subprocess.CompletedProcess:
        p=subprocess.run(self.base+list(args),capture_output=True,timeout=timeout)
        if check and p.returncode: raise ProbeError('ADB command failed: '+' '.join(args)+'\n'+p.stderr.decode(errors='replace'))
        return p
    def text(self,*args:str,check:bool=True)->str: return self.run(*args,check=check).stdout.decode(errors='replace').strip()

def fail(message: str) -> None: raise ProbeError(message)

def load_json(path: Path, missing: str) -> dict:
    try: value=json.loads(path.read_text())
    except FileNotFoundError: fail(missing)
    except (OSError, ValueError) as error: fail(f'Malformed {path.name}: {error}')
    if not isinstance(value,dict): fail(f'Malformed {path.name}')
    return value

def manifest_identity() -> dict:
    try:
        root=ET.parse(ROOT/'app/CastleEngineManifest.xml').getroot(); version=root.find('version')
        if root.tag!='project' or version is None: fail('Malformed permanent manifest')
        return {'name':root.attrib['name'],'package':root.attrib['qualified_name'],
                'version':version.attrib['value'],'version_code':int(version.attrib['code'])}
    except (OSError, ET.ParseError, KeyError, ValueError) as error: fail(f'Malformed permanent manifest: {error}')

def current_package() -> tuple[dict, str]:
    if not CURRENT_BUILD_METADATA.is_file(): raise RuntimeError('Current build metadata is missing')
    build=load_json(CURRENT_BUILD_METADATA,'Current build metadata is missing')
    if set(build)!={'schema','build_id','status','cpu','abi','application','upstream_commit','resource_count','compile','package','package_metadata','error'} or build.get('schema')!=CURRENT_BUILD_SCHEMA: fail('Invalid current build schema')
    if build.get('status')!='PASS': raise RuntimeError('Current build is not PASS')
    app=build.get('application'); identity=manifest_identity()
    if not isinstance(app,dict) or any(app.get(key)!=value for key,value in identity.items()): fail('Build application identity mismatch')
    if build.get('package_metadata')!='out/current/package.json': fail('Build package metadata locator mismatch')
    package=load_json(CURRENT_PACKAGE_METADATA,'Current package metadata is missing')
    if set(package)!={'schema','build_id','application','cpu','abi','apk'} or package.get('schema')!=CURRENT_PACKAGE_SCHEMA: fail('Invalid current package schema')
    if package.get('build_id')!=build.get('build_id') or package.get('application')!=identity: fail('Package metadata identity mismatch')
    if package.get('cpu')!=build.get('cpu') or package.get('abi')!=build.get('abi'): fail('Package build mismatch')
    apk=package.get('apk')
    if not isinstance(apk,dict) or set(apk)!={'file','sha256','size','native_library_path_in_apk'} or apk.get('file')!='out/current/kam-mobile.apk': fail('Package APK locator mismatch')
    if package['cpu'] not in CPUS or package['abi']!=CPUS[package['cpu']][0]: fail('Invalid package ABI')
    if not CURRENT_APK.is_file() or apk.get('sha256')!=sha256(CURRENT_APK): fail('Current APK missing or changed')
    return package, identity['package']

def main() -> int:
    ap=argparse.ArgumentParser(description=__doc__); ap.add_argument('--serial'); a=ap.parse_args()
    try:
        package,pkg=current_package()
    except ProbeError as error:
        print('FAIL\n'+str(error)); return 1
    except RuntimeError as error:
        print('BLOCKED\n'+str(error)); return 2
    except (KeyError, TypeError) as error:
        print('FAIL\n'+str(error)); return 1
    if not shutil.which('adb'):
        print('BLOCKED\nadb is not installed on host'); return 2
    try:
        devices=[line.split()[:2] for line in Adb(None).text('devices').splitlines()[1:] if line.strip()]
        if len(devices)!=1 or len(devices[0])!=2 or devices[0][1]!='device': fail('Require exactly one authorized device')
        serial=devices[0][0]
        if a.serial and a.serial!=serial: fail('Requested serial is not the sole authorized device')
        adb=Adb(serial)
        if package['abi'] not in adb.text('shell','getprop','ro.product.cpu.abilist').split(','): fail('Device ABI does not support package ABI')
    except (ProbeError, OSError, subprocess.TimeoutExpired) as error:
        print('BLOCKED\n'+str(error)); return 2
    try:
        install=adb.run('install','-r',str(CURRENT_APK),check=False)
        if install.returncode and b'INSTALL_FAILED_UPDATE_INCOMPATIBLE' in install.stdout+install.stderr:
            adb.run('uninstall',pkg,check=False); install=adb.run('install',str(CURRENT_APK),check=False)
        if install.returncode: fail('APK install failed')
        candidates=[line.strip() for line in adb.text('shell','cmd','package','resolve-activity','--brief','-a','android.intent.action.MAIN','-c','android.intent.category.LAUNCHER',pkg).splitlines() if line.strip()]
        if len(candidates)!=1 or not candidates[0].startswith(pkg+'/'): fail('Cannot resolve exactly one application launcher activity')
        if adb.run('shell','am','start','-W','-n',candidates[0],check=False).returncode: fail('Launch failed')
    except (ProbeError, OSError, subprocess.TimeoutExpired) as error:
        print('FAIL\n'+str(error)); return 1
    print('PASS'); return 0

if __name__=='__main__': raise SystemExit(main())