test_android_env_overrides.py

test_android_env_overrides.py

#!/usr/bin/env python3
"""ADB-based integration test for WiVRn runtime HMD env/sysprop overrides.

This script:
- sets Android debug sysprops used by WiVRn quirks,
- starts/stops the WiVRn Android activity,
- captures logcat,
- verifies override-presence and effective values from WiVRn logs.
"""

from __future__ import annotations

import argparse
import json
import subprocess
import sys
import time
from dataclasses import dataclass
from typing import Optional

PACKAGE_NAME = "org.meumeu.wivrn.local"
ACTIVITY_NAME = "org.meumeu.wivrn.MainActivity"
COMPONENT_NAME = f"{PACKAGE_NAME}/{ACTIVITY_NAME}"

PROP_PANEL = "debug.wivrn.panel_width_override"
PROP_OXR11 = "debug.wivrn.max_openxr_api_version"
PROP_SRGB = "debug.wivrn.needs_srgb_conversion"
PROP_CONTROLLER = "debug.wivrn.controller_profile"
UNSET_SENTINEL = "_"

INIT_HEADER = "HMD traits initialized"
TRAIT_FIELDS = {
    "controller_profile",
    "controller_ray_model",
    "panel_width_override",
    "max_openxr_api_version",
    "needs_srgb_conversion",
}


@dataclass(frozen=True)
class Case:
    name: str
    panel_width: Optional[str]
    disable_openxr_1_1: Optional[str]
    srgb: Optional[str]
    controller: Optional[str]


def run_adb(args: list[str], serial: Optional[str], check: bool = True) -> subprocess.CompletedProcess[str]:
    cmd = ["adb"]
    if serial:
        cmd += ["-s", serial]
    cmd += args
    proc = subprocess.run(cmd, text=True, capture_output=True, check=False)
    print(f"[ADB] {' '.join(cmd)}")
    const_is_logcat = len(args) > 0 and args[0] == "logcat"
    if proc.stdout.strip() and not const_is_logcat:
        print(f"[ADB:stdout]\n{proc.stdout.strip()}")
    if proc.stderr.strip():
        print(f"[ADB:stderr]\n{proc.stderr.strip()}")
    if check and proc.returncode != 0:
        raise subprocess.CalledProcessError(proc.returncode, cmd, output=proc.stdout, stderr=proc.stderr)
    return proc


def set_prop(serial: Optional[str], key: str, value: Optional[str]) -> None:
    if value is None:
        print(f"[INFO] Leaving {key} unchanged (None)")
        return
    if value == "":
        run_adb(["shell", "setprop", key, UNSET_SENTINEL], serial, check=False)
        return
    run_adb(["shell", "setprop", key, value], serial, check=False)


def clear_logcat(serial: Optional[str]) -> None:
    run_adb(["logcat", "-c"], serial)


def stop_app(serial: Optional[str], package_name: str) -> None:
    run_adb(["shell", "am", "force-stop", package_name], serial)


def start_app(serial: Optional[str], component_name: str) -> None:
    proc = run_adb(["shell", "am", "start", "-W", "-n", component_name], serial)
    if "Error:" in proc.stdout:
        raise RuntimeError(f"am start failed: {proc.stdout.strip()}")


def check_app_running(serial: Optional[str], package_name: str) -> bool:
    proc = run_adb(["shell", "pidof", package_name], serial, check=False)
    return bool(proc.stdout.strip())


def collect_logs(serial: Optional[str], timeout_s: float) -> str:
    deadline = time.time() + timeout_s
    collected = ""
    while time.time() < deadline:
        proc = run_adb(["logcat", "-d", "-v", "brief", "WiVRn:I", "*:S"], serial, check=False)
        chunk = proc.stdout
        if chunk:
            collected = f"{collected}\n{chunk}"
        if has_complete_trait_block(collected):
            return collected
        if chunk:
            print("[LOGCAT] Waiting for initialization logs...")
        time.sleep(0.5)
    return collected


def has_complete_trait_block(logs: str) -> bool:
    if INIT_HEADER not in logs:
        return False
    return all(
        f"{field}:" in logs or f"{field} override:" in logs
        for field in TRAIT_FIELDS
    )


def parse_flags(logs: str) -> tuple[dict[str, bool], dict[str, str]]:
    saw_header = False
    fields: dict[str, str] = {}
    flags = {
        "panel_width": False,
        "disable_openxr_1_1": False,
        "srgb_conversion": False,
        "controller_profile": False,
    }

    for raw_line in logs.splitlines():
        line = raw_line.strip()
        if INIT_HEADER in line:
            saw_header = True
            continue
        if not saw_header:
            continue

        for field_name in TRAIT_FIELDS:
            marker = f"{field_name} override:"
            plain = f"{field_name}:"
            if marker in line:
                value = line.split("->", 1)[1].strip()
                fields[field_name] = value
                if field_name == "panel_width_override":
                    flags["panel_width"] = True
                elif field_name == "max_openxr_api_version":
                    flags["disable_openxr_1_1"] = True
                elif field_name == "needs_srgb_conversion":
                    flags["srgb_conversion"] = True
                elif field_name == "controller_profile":
                    flags["controller_profile"] = True
                break
            if plain in line:
                value = line.split(plain, 1)[1].strip()
                fields.setdefault(field_name, value)
                break

    if not saw_header:
        raise AssertionError("Could not find 'HMD traits initialized' log line")

    missing = TRAIT_FIELDS - fields.keys()
    if missing:
        raise AssertionError(f"Could not find expected trait log lines: {sorted(missing)}")

    values = {
        "profile": fields["controller_profile"],
        "panel_width_override": fields["panel_width_override"],
        "max_openxr_api": fields["max_openxr_api_version"],
        "needs_srgb_conversion": fields["needs_srgb_conversion"],
    }
    return flags, values


def print_relevant_lines(logs: str) -> None:
    for line in logs.splitlines():
        if INIT_HEADER in line or any(f"{field}:" in line or f"{field} override:" in line for field in TRAIT_FIELDS):
            print(f"[MATCH] {line}")


def expected_presence(value: Optional[str], prop_key: str, baseline_profile: str) -> bool:
    del prop_key, baseline_profile
    return value is not None and value != ""


def run_case(case: Case, serial: Optional[str], timeout_s: float, package_name: str, component_name: str, restart_delay_s: float, baseline_profile: str) -> dict[str, str]:
    print(f"\n=== Running case: {case.name} ===")
    set_prop(serial, PROP_PANEL, case.panel_width)
    set_prop(serial, PROP_OXR11, case.disable_openxr_1_1)
    set_prop(serial, PROP_SRGB, case.srgb)
    set_prop(serial, PROP_CONTROLLER, case.controller)

    clear_logcat(serial)
    stop_app(serial, package_name)
    if restart_delay_s > 0:
        print(f"[WAIT] Sleeping {restart_delay_s:.1f}s before restart")
        time.sleep(restart_delay_s)
    start_app(serial, component_name)
    if not check_app_running(serial, package_name):
        raise RuntimeError("App did not start (pidof returned no PID)")
    logs = collect_logs(serial, timeout_s)
    print_relevant_lines(logs)

    flags, values = parse_flags(logs)

    assert flags["panel_width"] == expected_presence(case.panel_width, PROP_PANEL, baseline_profile), (
        f"{case.name}: panel_width override presence mismatch: got={flags['panel_width']} expected={expected_presence(case.panel_width, PROP_PANEL, baseline_profile)}")
    assert flags["disable_openxr_1_1"] == expected_presence(case.disable_openxr_1_1, PROP_OXR11, baseline_profile), (
        f"{case.name}: disable_openxr_1_1 override presence mismatch")
    assert flags["srgb_conversion"] == expected_presence(case.srgb, PROP_SRGB, baseline_profile), (
        f"{case.name}: srgb override presence mismatch")
    assert flags["controller_profile"] == expected_presence(case.controller, PROP_CONTROLLER, baseline_profile), (
        f"{case.name}: controller override presence mismatch")

    print(json.dumps({"case": case.name, "flags": flags, "values": values}, indent=2))
    return values


def build_cases() -> list[Case]:
    return [
        Case("all_unset", None, None, None, None),
        Case("all_empty", "", "", "", ""),
        Case("panel_only", "2048", "", "", ""),
        Case("disable_oxr11_only", "", "1.0", "", ""),
        Case("srgb_only", "", "", "false", ""),
        Case("controller_only", "", "", "", "custom-controller"),
        Case("all_set", "1832", "1.0", "on", "custom-controller"),
    ]


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--serial", help="adb device serial", default=None)
    parser.add_argument("--timeout", type=float, default=20.0, help="seconds to wait for expected logs")
    parser.add_argument("--restart-delay", type=float, default=2.0, help="seconds to wait between force-stop and start")
    parser.add_argument("--package", default=PACKAGE_NAME, help="Android package name to launch/stop")
    parser.add_argument("--component", default=COMPONENT_NAME, help="Android component name (<package>/<activity>)")
    args = parser.parse_args()

    cases = build_cases()
    failures = 0
    baseline_profile = ""
    for index, case in enumerate(cases):
        try:
            values = run_case(case, args.serial, args.timeout, args.package, args.component, args.restart_delay, baseline_profile)
            if index == 0:
                baseline_profile = values["profile"]
                print(f"[INFO] Baseline profile detected: {baseline_profile}")
        except Exception as exc:  # noqa: BLE001
            failures += 1
            print(f"[FAIL] {case.name}: {exc}")

    stop_app(args.serial, args.package)
    if failures:
        print(f"Completed with {failures} failure(s)")
        return 1
    print("All cases passed")
    return 0


if __name__ == "__main__":
    sys.exit(main())