"""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() 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())