147 lines
9.1 KiB
Python
147 lines
9.1 KiB
Python
"""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())
|