File size: 4,792 Bytes
0862f9d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 | """Shared helpers for QDC on-device test runners."""
from __future__ import annotations
import logging
import os
import subprocess
import tempfile
from appium.options.common import AppiumOptions
log = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# On-device paths
# ---------------------------------------------------------------------------
BUNDLE_PATH = "/data/local/tmp/llama.cpp"
BIN_PATH = f"{BUNDLE_PATH}/bin"
LIB_PATH = f"{BUNDLE_PATH}/lib"
QDC_LOGS_PATH = "/data/local/tmp/QDC_logs"
SCRIPTS_DIR = "/qdc/appium"
MODEL_NAME = "model.gguf"
MODEL_DEVICE_PATH = "/data/local/tmp/gguf/model.gguf"
PROMPT_DIR = "/data/local/tmp/scorecard_prompts"
# ---------------------------------------------------------------------------
# Appium session options
# ---------------------------------------------------------------------------
options = AppiumOptions()
options.set_capability("automationName", "UiAutomator2")
options.set_capability("platformName", "Android")
options.set_capability("deviceName", os.getenv("ANDROID_DEVICE_VERSION"))
# ---------------------------------------------------------------------------
# Shell / process helpers
# ---------------------------------------------------------------------------
def write_qdc_log(filename: str, content: str) -> None:
"""Write content as a log file for QDC log collection."""
subprocess.run(
["adb", "shell", f"mkdir -p {QDC_LOGS_PATH}"],
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
)
with tempfile.NamedTemporaryFile(mode="w", suffix=".log", delete=False) as f:
f.write(content)
tmp_path = f.name
try:
subprocess.run(
["adb", "push", tmp_path, f"{QDC_LOGS_PATH}/{filename}"],
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
)
finally:
os.unlink(tmp_path)
def ensure_bundle(check_binary: str | None = None) -> None:
"""Ensure the llama_cpp_bundle is available on the target device."""
push_bundle_if_needed(check_binary or f"{BIN_PATH}/llama-cli")
# ---------------------------------------------------------------------------
# Android / Linux host helpers
# ---------------------------------------------------------------------------
def run_adb_command(cmd: str, *, check: bool = True) -> subprocess.CompletedProcess:
"""Run a command on-device via ``adb shell`` with exit-code sentinel."""
raw = subprocess.run(
["adb", "shell", f"{cmd}; echo __RC__:$?"],
text=True,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
)
stdout = raw.stdout
returncode = raw.returncode
if stdout:
lines = stdout.rstrip("\n").split("\n")
if lines and lines[-1].startswith("__RC__:"):
try:
returncode = int(lines[-1][7:])
stdout = "\n".join(lines[:-1]) + "\n"
except ValueError:
pass
log.info(stdout)
result = subprocess.CompletedProcess(raw.args, returncode, stdout=stdout)
if check:
assert returncode == 0, f"Command failed (exit {returncode})"
return result
def run_script(
script: str,
extra_env: dict[str, str] | None = None,
extra_args: list[str] | None = None,
) -> subprocess.CompletedProcess:
"""Run an upstream shell script from /qdc/appium/ on the QDC runner host."""
env = os.environ.copy()
env["GGML_HEXAGON_EXPERIMENTAL"] = "1"
if extra_env:
env.update(extra_env)
cmd = [f"{SCRIPTS_DIR}/{script}"] + (extra_args or [])
result = subprocess.run(
cmd, env=env,
text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
)
log.info(result.stdout)
return result
def adb_shell(cmd: str) -> None:
"""Run a command via adb shell (fire-and-forget, no error check)."""
subprocess.run(
["adb", "shell", "sh", "-c", cmd],
capture_output=True, encoding="utf-8", errors="replace", check=False,
)
def push_bundle_if_needed(check_binary: str) -> None:
"""Push llama_cpp_bundle to the device if check_binary is not already present."""
result = subprocess.run(
["adb", "shell", f"ls {check_binary}"],
text=True,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
)
if result.returncode != 0:
subprocess.run(
["adb", "push", "/qdc/appium/llama_cpp_bundle/", BUNDLE_PATH],
text=True,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
)
subprocess.run(
["adb", "shell", f"find {BUNDLE_PATH}/bin -type f -exec chmod 755 {{}} +"],
text=True,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
)
|