kam-mobile/tools/device.py

"""Install, launch, and capture startup diagnostics for the current APK."""
from __future__ import annotations
import argparse, json, os, shutil, subprocess, time, 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)

AVD_NAME='kam_mobile_api30'
BOOT_TIMEOUT=180

class Adb:
    def __init__(self, serial: str | None): self.base=[shutil.which('adb') or '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,timeout:int=120)->str: return self.run(*args,check=check,timeout=timeout).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 apk.get('native_library_path_in_apk')!=f"lib/{package['abi']}/libkam_mobile_android.so": fail('Package native library path does not match selected 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 authorized_devices() -> list[str]:
    return [line.split()[0] for line in Adb(None).text('devices').splitlines()[1:]
            if len(line.split()) >= 2 and line.split()[1] == 'device']

def start_avd() -> str:
    emulator=shutil.which('emulator')
    if not emulator: fail('Android Emulator is not installed on host')
    if AVD_NAME not in subprocess.run([emulator,'-list-avds'],capture_output=True,text=True,timeout=30).stdout.splitlines():
        fail(f'Android AVD {AVD_NAME} is not available')
    accel='on' if subprocess.run([emulator,'-accel-check'],capture_output=True,timeout=30).returncode==0 else 'off'
    subprocess.Popen([emulator,'-avd',AVD_NAME,'-no-window','-no-audio','-no-boot-anim',
                      '-gpu','swiftshader_indirect','-accel',accel,'-no-snapshot-save'],
                     stdout=subprocess.DEVNULL,stderr=subprocess.DEVNULL,start_new_session=True,
                     env={**os.environ,'ANDROID_EMULATOR_WAIT_TIME_BEFORE_KILL':'180'})
    Adb(None).run('wait-for-device',timeout=BOOT_TIMEOUT)
    deadline=time.monotonic()+BOOT_TIMEOUT
    while time.monotonic()<deadline:
        devices=authorized_devices()
        if len(devices)==1:
            adb=Adb(devices[0])
            if adb.text('shell','getprop','sys.boot_completed',check=False,timeout=10)=='1': return devices[0]
        time.sleep(2)
    fail(f'Android AVD {AVD_NAME} did not boot within {BOOT_TIMEOUT} seconds')

def choose_device(requested: str | None) -> str:
    devices=authorized_devices()
    if requested:
        if requested in devices: return requested
        if devices: fail('Requested serial is not authorized')
    if not devices:
        serial=start_avd()
        if requested and serial!=requested: fail('Requested serial is not the prepared emulator')
        return serial
    if len(devices)!=1: fail('Require exactly one authorized device or use --serial')
    return devices[0]

def write_runtime_device(path: Path, **values: str) -> None:
    path.parent.mkdir(parents=True,exist_ok=True)
    path.write_text(''.join(f'{key}: {value}\n' for key,value in values.items()),encoding='utf-8')

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:
        serial=choose_device(a.serial); adb=Adb(serial)
        abi=adb.text('shell','getprop','ro.product.cpu.abi')
        abilist=adb.text('shell','getprop','ro.product.cpu.abilist')
        android=adb.text('shell','getprop','ro.build.version.release')+'/'+adb.text('shell','getprop','ro.build.version.sdk')
        if package['abi'] not in abilist.split(','): fail('Device ABI does not support package ABI')
    except (ProbeError, OSError, subprocess.TimeoutExpired) as error:
        print('BUILD PASS\nRUNTIME BLOCKED\n'+str(error)); return 2
    logs=ROOT/'build-evidence'/package['build_id']/'logs'
    logcat=logs/'runtime-logcat.txt'; device=logs/'runtime-device.txt'
    launcher=''; pid=''; result='RUNTIME FAIL'
    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: '+(install.stdout+install.stderr).decode(errors='replace').strip())
        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().startswith(pkg+'/')]
        if len(candidates)!=1: fail('Cannot resolve exactly one application launcher activity')
        launcher=candidates[0]
        adb.run('shell','am','force-stop',pkg)
        adb.run('logcat','-c')
        if adb.run('shell','am','start','-W','-n',launcher,check=False).returncode: fail('Launch failed')
        time.sleep(5)
        pid=adb.text('shell','pidof',pkg,check=False)
        result='RUNTIME PASS' if pid else 'RUNTIME FAIL'
    except (ProbeError, OSError, subprocess.TimeoutExpired) as error:
        result='RUNTIME FAIL: '+str(error)
    finally:
        logs.mkdir(parents=True,exist_ok=True)
        try: captured=adb.run('logcat','-d','-v','threadtime',check=False).stdout.decode(errors='replace')
        except (OSError, subprocess.TimeoutExpired) as error: captured='Could not capture logcat: '+str(error)+'\n'
        logcat.write_text(captured,encoding='utf-8')
        write_runtime_device(device,serial=serial,android=android,ro_product_cpu_abi=abi,
                             ro_product_cpu_abilist=abilist,launcher_activity=launcher,
                             pid_after_startup_wait=pid,runtime_result=result)
    print('BUILD PASS\n'+result+'\nlogcat: '+str(logcat.relative_to(ROOT)))
    return 0 if result=='RUNTIME PASS' else 1

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