From 133713377a0a4abccca1618a665a24da07806202 Mon Sep 17 00:00:00 2001 From: Billy Lau Date: Mon, 28 Sep 2026 22:04:34 -0500 Subject: [PATCH] automate_observation: add --events for machine-readable progress Add an opt-in --events PATH flag ("-" for stdout) that writes progress as JSON Lines, so CI jobs and other wrappers need not scrape logs. Events: run_started, devices, device_started, step, device_finished, run_finished. run_finished comes last even on early exits, exceptions, Ctrl-C and SIGTERM (caught, then re-raised; a stream without it means failure, e.g. SIGKILL). It has `ok` and a per-device summary. Device status is success, partial_check_incomplete, partial_error or failed; success means the inclusion proof check completed, not that every split verified. Errors are {reason, message} with stable reason codes. With "-", other stdout output goes to stderr. Without --events, behaviour is unchanged, except log lines from the old main() now show run() as their function name. Schema in docs/automate_observation.md; additive changes keep v=1. Test: - pytest uraniborg/scripts/python/tests/: 116 passed (47 new, 69 existing unmodified). - Live on an Android 15 emulator: --events -, inclusion proof check, unknown --serial, SIGTERM mid-run, and no --events (output unchanged). Change-Id: I04e40b0de40d864aac92ac7bcf466a6ccf606b61 --- uraniborg/docs/automate_observation.md | 161 ++++ .../scripts/python/automate_observation.py | 693 +++++++++++--- .../tests/test_automate_observation_events.py | 879 ++++++++++++++++++ 3 files changed, 1612 insertions(+), 121 deletions(-) create mode 100644 uraniborg/scripts/python/tests/test_automate_observation_events.py diff --git a/uraniborg/docs/automate_observation.md b/uraniborg/docs/automate_observation.md index 2c62914..82f7d84 100644 --- a/uraniborg/docs/automate_observation.md +++ b/uraniborg/docs/automate_observation.md @@ -42,6 +42,167 @@ A requested serial that is not connected is not silently skipped: it is logged as an error, reported as `FAILED` in the final summary, and makes the script exit with code `1`. The remaining requested devices are still observed. +### Machine-Readable Progress (`--events`) + +Log messages are meant for people and may change between versions. Wrappers +such as CI jobs or device-lab scripts should use `--events` instead: it writes +a stream of progress events as [JSON Lines](https://jsonlines.org/), one JSON +object per line, flushed as each event happens. + +```bash +# Write events to a file (overwritten if it exists): +python3 automate_observation.py --events run.jsonl + +# Or stream them on stdout: +python3 automate_observation.py --events - | my-wrapper +``` + +With `--events -`, stdout carries **only** events. Everything else that would +otherwise have gone to stdout is sent to stderr, including output from child +processes such as the verifier and the Xiaomi manual-install prompt. Logging +always goes to stderr. Without `--events`, nothing changes: log output, exit +codes and the results layout are the same. If the `--events` destination +cannot be opened, the script exits with code `1` before doing anything else. +If writing events fails partway through (for example, the reading process +exits), a warning is logged and the run continues without events. + +#### Event schema (version 1) + +Every event has these fields: + +| Field | Meaning | +| :--- | :--- | +| `v` | Schema version, currently `1` | +| `ts` | UTC timestamp, ISO 8601 with milliseconds, e.g. `2026-01-02T03:04:05.678Z` | +| `type` | One of the event types below | + +Optional fields are left out rather than set to `null`. + +| `type` | Fields | When | +| :--- | :--- | :--- | +| `run_started` | `argv` (arguments, without the script name), `pid` | First event of every run | +| `step` | `step`, `state` (`started` / `finished` / `failed`), `device`\*, `duration_ms`\*\*, `message`\* | Around each phase (see below) | +| `devices` | `devices` (list of `{serial, unauthorized, model?, product?, device?}`), `selected` (serials to observe), `missing` (requested via `--serial` but not connected) | Once, after listing devices | +| `device_started` | `device` | Before processing each selected device | +| `device_finished` | `device`, `status`, `results_dir`\*, `error`\* | After each selected device, and once for each missing serial | +| `run_finished` | `exit_code`, `ok`, `summary`, `error`\* | Last event of a run that ends normally, exits early, raises, or is stopped by Ctrl-C or `SIGTERM` (see [End of stream](#end-of-stream)) | + +\* only when relevant. \*\* only on `finished` / `failed`. + +Both `error` fields are `{reason, message}`: `reason` is a stable code (see +[Error reasons](#error-reasons)) to match on, and `message` is human-readable +text that may change. Step `message` is free text. + +**Steps.** Run-level steps have no `device` field and run in this order: +`build_hubble` (only without `-H`), `verify_hubble`, `check_adb`, +`start_adb_server`, `list_devices`. Per-device steps carry `device`: +`uninstall_previous` (only if Hubble was already installed), `install_hubble`, +`launch_hubble`, `wait_for_results`, `extract_results`, `extract_selinux`, and, +with `--perform_inclusion_proof_check`, `inclusion_proof_prefetch` (only when +pre-fetching is attempted) and `inclusion_proof_check`. A failed +`inclusion_proof_prefetch` is not fatal: verification falls back to fetching +entries on demand. + +**Device status.** `device_finished.status` and `summary[serial].status` use the +same four outcomes as the final log summary: + +| `status` | Meaning | +| :--- | :--- | +| `success` | Results collected, and the inclusion proof check (if requested) completed | +| `partial_check_incomplete` | Results collected, but the inclusion proof check could not complete (for example, the package list was missing or unreadable, or the output could not be written) | +| `partial_error` | Results collected, but an unexpected error occurred afterwards, or the run was stopped (`error.reason` is `interrupted` or `terminated`) | +| `failed` | No results collected (including unauthorized and not-connected devices, and devices stopped before results were collected) | + +> [!IMPORTANT] +> `success` does **not** mean every APK was found in the transparency log. A +> completed check records a per-split `inclusion_proof_verified` result, which +> may be `false`, in `packages_with_inclusion_proof_signal.txt` (or +> `preinstalled_packages_with_inclusion_proof_signal.txt`) inside +> `results_dir`. Read that file to decide whether the APKs are in the log. + +`results_dir` is present whenever results were collected, including both +`partial_*` outcomes. + +**Run outcome.** `run_finished.summary` maps each serial, in the order it was +requested (or discovered, without `--serial`), to `{status, results_dir?}`. +If the run is stopped early (`interrupted`, `terminated` or +`unexpected_error`), `summary` lists only the devices that reached +`device_finished`; devices that had not been processed yet are absent. +`ok` is `true` only when `exit_code` is `0` **and** there is no `error`. Treat +`ok`, not `exit_code`, as the success signal: when the script stops before +processing any device (for example, no device is connected), it still exits +with code `0`, but `run_finished.error` explains why. + +#### Error reasons + +`run_finished.error.reason`: + +| `reason` | Meaning | +| :--- | :--- | +| `unsupported_platform` | The host OS is not supported | +| `android_sdk_not_found` | Rebuilding Hubble (no `-H`) failed: Android SDK not found | +| `hubble_build_failed` | Rebuilding Hubble failed (Gradle or symlink error) | +| `invalid_hubble_apk` | The `-H` path is not an `.apk` | +| `adb_not_found` | `adb` is not installed | +| `adb_server_failed` | The adb server could not be started | +| `adb_devices_failed` | `adb devices` failed (without `--serial`) | +| `no_devices` | No device is connected (without `--serial`) | +| `unexpected_error` | An unhandled exception; `exit_code` is `1` | +| `interrupted` | Interrupted with Ctrl-C; `exit_code` is `130` | +| `terminated` | Stopped by `SIGTERM`; `exit_code` is `143` | + +`device_finished.error.reason`: + +| `reason` | Meaning | `status` | +| :--- | :--- | :--- | +| `not_connected` | Requested via `--serial` but not connected | `failed` | +| `unauthorized` | ADB is not authorized on the device | `failed` | +| `uninstall_failed` | A previous Hubble installation could not be removed | `failed` | +| `install_failed` | Hubble could not be installed | `failed` | +| `launch_failed` | Hubble could not be launched | `failed` | +| `no_results` | Hubble produced no results in time | `failed` | +| `extract_failed` | Results could not be pulled from the device | `failed` | +| `inclusion_proof_check_incomplete` | The inclusion proof check could not complete | `partial_check_incomplete` | +| `unexpected_error` | An unhandled exception while processing the device | `failed` or `partial_error` | +| `interrupted` | Ctrl-C while processing the device | `failed` or `partial_error` | +| `terminated` | `SIGTERM` while processing the device | `failed` or `partial_error` | + +For the last three, `status` is `partial_error` if results had already been +collected, otherwise `failed`. + +Example, for one device with inclusion proofs: + +```json +{"v": 1, "ts": "2026-01-02T03:04:05.678Z", "type": "run_started", "argv": ["--events", "-", "--perform_inclusion_proof_check", "--verifier_path=/path/to/verifier"], "pid": 4242} +{"v": 1, "ts": "2026-01-02T03:04:05.690Z", "type": "step", "step": "verify_hubble", "state": "started"} +{"v": 1, "ts": "2026-01-02T03:04:05.691Z", "type": "step", "step": "verify_hubble", "state": "finished", "duration_ms": 1} +... +{"v": 1, "ts": "2026-01-02T03:04:06.100Z", "type": "devices", "devices": [{"serial": "ABCDEF012345", "unauthorized": false, "model": "Pixel_9"}], "selected": ["ABCDEF012345"], "missing": []} +{"v": 1, "ts": "2026-01-02T03:04:06.101Z", "type": "device_started", "device": "ABCDEF012345"} +{"v": 1, "ts": "2026-01-02T03:04:06.102Z", "type": "step", "step": "install_hubble", "state": "started", "device": "ABCDEF012345"} +... +{"v": 1, "ts": "2026-01-02T03:09:41.310Z", "type": "device_finished", "device": "ABCDEF012345", "status": "success", "results_dir": "/path/to/results/google/.../000"} +{"v": 1, "ts": "2026-01-02T03:09:41.312Z", "type": "run_finished", "exit_code": 0, "ok": true, "summary": {"ABCDEF012345": {"status": "success", "results_dir": "/path/to/results/google/.../000"}}} +``` + +#### End of stream + +`run_finished` cannot be written if the process is killed with `SIGKILL`, the +Python interpreter crashes, or the host goes down. **A stream that ends without +`run_finished` means the run failed.** Do not infer success from the +`device_finished` events seen so far. + +With `--events`, `SIGTERM` (how CI systems usually enforce timeouts and +cancellations) is caught so that `run_finished` can be written. The script then +still terminates by `SIGTERM`, so the parent process sees the same exit status +as before. Without `--events`, `SIGTERM` handling is unchanged. + +#### Compatibility + +The schema is a stable contract. New event types and new fields may be added +without changing `v`, so consumers should ignore types and fields they do not +recognize. Removing a field or changing what it means requires a new `v`. + ## Performing Inclusion Proof Checks You can automatically verify extracted package APK splits against Android Binary diff --git a/uraniborg/scripts/python/automate_observation.py b/uraniborg/scripts/python/automate_observation.py index 0de14f9..1ec55fb 100644 --- a/uraniborg/scripts/python/automate_observation.py +++ b/uraniborg/scripts/python/automate_observation.py @@ -21,15 +21,19 @@ Hubble data back to host. """ import argparse +import contextlib +import datetime import io # to convert regular buffer to in-memory bytes buffer for tarfileobj import json import logging import os import shutil # to help move files +import signal import sys import tarfile # to untar decompressed adb backup file import tempfile -from typing import Optional # runtime support for type hints. +import time +from typing import Optional, TextIO # runtime support for type hints. import zlib # to decompress adb backup import inclusion_proof_check @@ -76,6 +80,15 @@ def parse_arguments() -> argparse.Namespace: "the order given. If omitted, every connected " "device is observed. A requested serial that is " "not connected is reported as FAILED.") + parser.add_argument("--events", required=False, default=None, + metavar="PATH", + help="If specified, writes machine-readable progress " + "events as JSON Lines (one JSON object per line) " + "to PATH, overwriting it. Use \"-\" to write events " + "to stdout; everything else the script would print " + "to stdout is then sent to stderr, so stdout " + "carries only events. See " + "docs/automate_observation.md for the schema.") parser.add_argument("--pull-all-apks", required=False, action="count", help="If specified, the script will attempt to download " @@ -157,6 +170,249 @@ def set_up_logging(args: argparse.Namespace) -> logging.Logger: return logger +# --- Machine-readable progress events (--events) ------------------------------ +# +# The schema is documented in docs/automate_observation.md and is a stable +# contract: fields and event types may be added without bumping +# EVENTS_SCHEMA_VERSION, but never removed or changed in meaning. + +EVENTS_SCHEMA_VERSION = 1 + +# Per-device outcomes. These are the same four outcomes the final log summary +# distinguishes. +STATUS_SUCCESS = "success" +STATUS_PARTIAL_CHECK_INCOMPLETE = "partial_check_incomplete" +STATUS_PARTIAL_ERROR = "partial_error" +STATUS_FAILED = "failed" + +# Error reasons. `error` fields in run_finished and device_finished are +# {reason, message}: reason is one of these stable codes, message is free text. +# +# run_finished.error, when the run stops before processing any device: +REASON_UNSUPPORTED_PLATFORM = "unsupported_platform" +REASON_ANDROID_SDK_NOT_FOUND = "android_sdk_not_found" +REASON_HUBBLE_BUILD_FAILED = "hubble_build_failed" +REASON_INVALID_HUBBLE_APK = "invalid_hubble_apk" +REASON_ADB_NOT_FOUND = "adb_not_found" +REASON_ADB_SERVER_FAILED = "adb_server_failed" +REASON_ADB_DEVICES_FAILED = "adb_devices_failed" +REASON_NO_DEVICES = "no_devices" +# device_finished.error only: +REASON_NOT_CONNECTED = "not_connected" +REASON_UNAUTHORIZED = "unauthorized" +REASON_UNINSTALL_FAILED = "uninstall_failed" +REASON_INSTALL_FAILED = "install_failed" +REASON_LAUNCH_FAILED = "launch_failed" +REASON_NO_RESULTS = "no_results" +REASON_EXTRACT_FAILED = "extract_failed" +REASON_INCLUSION_PROOF_CHECK_INCOMPLETE = "inclusion_proof_check_incomplete" +# Both run_finished.error and device_finished.error: +REASON_UNEXPECTED_ERROR = "unexpected_error" +REASON_INTERRUPTED = "interrupted" +REASON_TERMINATED = "terminated" + + +class Terminated(BaseException): + """Raised by the SIGTERM handler that main() installs while --events is on. + + Derives from BaseException, like KeyboardInterrupt, so that the generic + `except Exception` handlers do not swallow it. SyscallWrapper only catches + KeyboardInterrupt, so this also propagates out of adb calls. + """ + + def __init__(self): + super().__init__("received SIGTERM") + + +def _raise_terminated(signum, frame): # pylint: disable=unused-argument + # Ignore repeated SIGTERMs while cleaning up so that run_finished is still + # written; _die_by_sigterm() restores the default action afterwards. + signal.signal(signal.SIGTERM, signal.SIG_IGN) + raise Terminated() + + +def _interruption_error(e: BaseException) -> dict: + """Describes an exception that is not an Exception, for device errors.""" + if isinstance(e, KeyboardInterrupt): + return _error(REASON_INTERRUPTED, "Interrupted.") + if isinstance(e, Terminated): + return _error(REASON_TERMINATED, "Terminated.") + return _error(REASON_UNEXPECTED_ERROR, + "{}: {}".format(type(e).__name__, e)) + + +def open_event_stream(path: str) -> TextIO: + """Opens the destination for --events. + + For "-", events go to the process's original stdout. To guarantee that + stdout then carries nothing but events, file descriptor 1 is redirected to + stderr afterwards, so anything else written to stdout -- by this script, + input() prompts, or child processes that inherit stdout -- lands on stderr. + + Args: + path: A file path, or "-" for stdout. + + Returns: + A writable text stream. + + Raises: + OSError: if the destination cannot be opened. + """ + if path == "-": + sys.stdout.flush() + events_fd = os.dup(sys.stdout.fileno()) + os.dup2(sys.stderr.fileno(), sys.stdout.fileno()) + return os.fdopen(events_fd, "w", encoding="utf-8") + return open(path, "w", encoding="utf-8") + + +class _Step: + """Handle yielded by EventEmitter.step() to mark a step as failed.""" + + def __init__(self): + self.failed = False + self.message = None + + def fail(self, message: str) -> str: + """Marks the step failed and returns message, for convenient reuse.""" + self.failed = True + self.message = message + return message + + +class EventEmitter: + """Writes progress events as JSON Lines. + + Without a stream every method is a no-op, so callers never need to check + whether --events was given. Writing is best-effort: if the destination + breaks (e.g. the reading process went away), a warning is logged once and + the run carries on without events. + """ + + def __init__(self, stream: Optional[TextIO] = None, + logger: Optional[logging.Logger] = None, + clock=None): + self._stream = stream + self._logger = logger + self._clock = clock or ( + lambda: datetime.datetime.now(datetime.timezone.utc)) + self._run_finished = False + # serial -> {status, results_dir?}, in device_finished order. + self._finished_devices = {} + + @property + def enabled(self) -> bool: + return self._stream is not None + + def emit(self, event_type: str, **fields): + """Writes one event. Fields whose value is None are omitted.""" + if self._stream is None: + return + timestamp = self._clock().isoformat(timespec="milliseconds") + event = {"v": EVENTS_SCHEMA_VERSION, + "ts": timestamp.replace("+00:00", "Z"), + "type": event_type} + event.update({k: v for k, v in fields.items() if v is not None}) + try: + self._stream.write(json.dumps(event, default=str) + "\n") + self._stream.flush() + except (OSError, ValueError) as e: + # ValueError: write to a closed file. + if self._logger: + self._logger.warning("Failed to write --events output (%s); " + "continuing without events.", e) + self._stream = None + + def device_finished(self, device: str, status: str, + results_dir: Optional[str] = None, + error: Optional[dict] = None): + """Emits device_finished and remembers the outcome for finish_run().""" + entry = {"status": status} + if results_dir is not None: + entry["results_dir"] = results_dir + self._finished_devices[device] = entry + self.emit("device_finished", device=device, status=status, + results_dir=results_dir, error=error) + + @contextlib.contextmanager + def step(self, name: str, device: Optional[str] = None): + """Brackets a phase with step started/finished/failed events. + + The step is reported as failed if the body raises (the exception is + re-raised) or calls fail() on the yielded handle; otherwise as finished. + """ + handle = _Step() + start = time.monotonic() + self.emit("step", step=name, state="started", device=device) + try: + yield handle + except BaseException as e: + self.emit("step", step=name, state="failed", device=device, + duration_ms=int((time.monotonic() - start) * 1000), + message="{}: {}".format(type(e).__name__, e)) + raise + self.emit("step", step=name, + state="failed" if handle.failed else "finished", + device=device, + duration_ms=int((time.monotonic() - start) * 1000), + message=handle.message) + + def finish_run(self, exit_code: int, summary: Optional[dict] = None, + error: Optional[dict] = None): + """Emits run_finished. Only the first call has any effect. + + If summary is None, it defaults to the devices reported through + device_finished() so far, so a run that is cut short still reports the + devices that did finish. + """ + if self._run_finished: + return + self._run_finished = True + self.emit("run_finished", + exit_code=exit_code, + ok=(exit_code == 0 and error is None), + summary=(summary if summary is not None + else dict(self._finished_devices)), + error=error) + + def close(self): + if self._stream is not None: + try: + self._stream.close() + except OSError: + pass + self._stream = None + + +def _error(reason: str, message: str) -> dict: + """Builds an `error` field: a stable REASON_* code plus free text.""" + return {"reason": reason, "message": message} + + +def _device_to_event(device) -> dict: + """Describes a DeviceInfo for the devices event.""" + info = {"serial": device.serial_number, + "unauthorized": bool(device.unauthorized)} + for key, attr in (("model", "model_name"), ("product", "product_name"), + ("device", "device_name")): + value = getattr(device, attr, None) + if isinstance(value, str) and value: + info[key] = value + return info + + +def device_status(serial: str, results: dict, missing_serials, + collection_error_devices, verification_failed_devices) -> str: + """Classifies a device's outcome into one of the STATUS_* values.""" + if serial in missing_serials or serial not in results: + return STATUS_FAILED + if serial in collection_error_devices: + return STATUS_PARTIAL_ERROR + if serial in verification_failed_devices: + return STATUS_PARTIAL_CHECK_INCOMPLETE + return STATUS_SUCCESS + + def supported_platform(logger: logging.Logger) -> bool: """Checks if this script is running on supported platform. @@ -891,69 +1147,131 @@ def ensure_android_sdk(hubble_project_dir: str, logger: logging.Logger) -> bool: return True -def main(): - args = parse_arguments() - logger = set_up_logging(args) +def rebuild_hubble(logger: logging.Logger) -> tuple[Optional[str], Optional[dict]]: + """Rebuilds the Hubble debug APK with Gradle and refreshes prebuilts/APK/latest. - if not supported_platform(logger): - logger.error("Sorry, your OS is currently unsupported for this script.") - return + Args: + logger: A logger object to log debug or error messages. - if args.hubble is None: - logger.info("-H flag not used. Rebuilding Hubble...") - script_dir = os.path.dirname(os.path.abspath(__file__)) - hubble_project_dir = os.path.abspath(os.path.join(script_dir, "../../AndroidStudioProject/Hubble")) - if not ensure_android_sdk(hubble_project_dir, logger): - logger.error("Failed to (re)build Hubble APK: Android SDK not found.") - return - gradlew_path = os.path.join(hubble_project_dir, "gradlew") + Returns: + A tuple (apk_path, error). On success apk_path is the path of the + refreshed "latest" symlink and error is None. On failure apk_path is None + and error is a run_finished error dict (see _error). + """ + script_dir = os.path.dirname(os.path.abspath(__file__)) + hubble_project_dir = os.path.abspath(os.path.join(script_dir, "../../AndroidStudioProject/Hubble")) + if not ensure_android_sdk(hubble_project_dir, logger): + msg = "Failed to (re)build Hubble APK: Android SDK not found." + logger.error(msg) + return None, _error(REASON_ANDROID_SDK_NOT_FOUND, msg) + gradlew_path = os.path.join(hubble_project_dir, "gradlew") - logger.info("Running 'gradlew assembleDebug' in %s", hubble_project_dir) - sw = SyscallWrapper(logger) - sw.call_returnable_command([gradlew_path, "assembleDebug"], cwd=hubble_project_dir) - if sw.error_occured: - logger.error("Failed to (re)build Hubble APK: [%d] %s", sw.return_code, sw.error_message) - return + logger.info("Running 'gradlew assembleDebug' in %s", hubble_project_dir) + sw = SyscallWrapper(logger) + sw.call_returnable_command([gradlew_path, "assembleDebug"], cwd=hubble_project_dir) + if sw.error_occured: + logger.error("Failed to (re)build Hubble APK: [%d] %s", sw.return_code, sw.error_message) + return None, _error( + REASON_HUBBLE_BUILD_FAILED, + "Failed to (re)build Hubble APK: [{}] {}".format( + sw.return_code, sw.error_message)) + + for line in sw.result_final: + logger.debug("Gradle output: %s", line) + + latest_symlink_path = os.path.abspath(os.path.join(script_dir, "../../prebuilts/APK/latest")) + os.makedirs(os.path.dirname(latest_symlink_path), exist_ok=True) + if os.path.exists(latest_symlink_path) or os.path.islink(latest_symlink_path): + logger.debug("Removing old symlink: %s", latest_symlink_path) + try: + os.remove(latest_symlink_path) + except Exception as e: + logger.error("Failed to remove old symlink %s: %s", latest_symlink_path, e) + return None, _error( + REASON_HUBBLE_BUILD_FAILED, + "Failed to remove old symlink {}: {}".format(latest_symlink_path, e)) - for line in sw.result_final: - logger.debug("Gradle output: %s", line) + symlink_target = "../../AndroidStudioProject/Hubble/app/build/outputs/apk/debug/app-debug.apk" - latest_symlink_path = os.path.abspath(os.path.join(script_dir, "../../prebuilts/APK/latest")) - os.makedirs(os.path.dirname(latest_symlink_path), exist_ok=True) - if os.path.exists(latest_symlink_path) or os.path.islink(latest_symlink_path): - logger.debug("Removing old symlink: %s", latest_symlink_path) - try: - os.remove(latest_symlink_path) - except Exception as e: - logger.error("Failed to remove old symlink %s: %s", latest_symlink_path, e) - return + logger.info("Creating new symlink 'latest' -> %s", symlink_target) + try: + os.symlink(symlink_target, latest_symlink_path) + except Exception as e: + logger.error("Failed to create symlink %s: %s", latest_symlink_path, e) + return None, _error( + REASON_HUBBLE_BUILD_FAILED, + "Failed to create symlink {}: {}".format(latest_symlink_path, e)) - symlink_target = "../../AndroidStudioProject/Hubble/app/build/outputs/apk/debug/app-debug.apk" + return latest_symlink_path, None - logger.info("Creating new symlink 'latest' -> %s", symlink_target) - try: - os.symlink(symlink_target, latest_symlink_path) - except Exception as e: - logger.error("Failed to create symlink %s: %s", latest_symlink_path, e) - return - args.hubble = latest_symlink_path +def run(args: argparse.Namespace, logger: logging.Logger, + events: EventEmitter) -> int: + """Runs the observation workflow. - if not verify_hubble(args, logger): - return + Emits run_finished on every path that returns. - if not adb_installed(logger): - return + Args: + args: Parsed arguments from parse_arguments(). + logger: A logger object to log debug or error messages. + events: Where to report progress events. May be a disabled emitter. - if not AdbWrapper.start_server(logger): - return + Returns: + The process exit code. Early exits (before any device is processed) + return 0, as they always have; run_finished.error describes them. + """ + events.emit("run_started", argv=sys.argv[1:], pid=os.getpid()) + + def early_exit(reason: str, message: str) -> int: + events.finish_run(0, error=_error(reason, message)) + return 0 + + if not supported_platform(logger): + logger.error("Sorry, your OS is currently unsupported for this script.") + return early_exit(REASON_UNSUPPORTED_PLATFORM, + "Unsupported platform: {}".format(sys.platform)) - connected_devices = AdbWrapper.devices(logger) or [] + if args.hubble is None: + logger.info("-H flag not used. Rebuilding Hubble...") + with events.step("build_hubble") as s: + hubble_path, build_error = rebuild_hubble(logger) + if build_error: + s.fail(build_error["message"]) + if build_error: + events.finish_run(0, error=build_error) + return 0 + args.hubble = hubble_path + + with events.step("verify_hubble") as s: + if not verify_hubble(args, logger): + s.fail("Hubble APK to be installed must have the .apk extension.") + if s.failed: + return early_exit(REASON_INVALID_HUBBLE_APK, s.message) + + with events.step("check_adb") as s: + if not adb_installed(logger): + s.fail("adb was not found on this system.") + if s.failed: + return early_exit(REASON_ADB_NOT_FOUND, s.message) + + with events.step("start_adb_server") as s: + if not AdbWrapper.start_server(logger): + s.fail("Failed to start the adb server.") + if s.failed: + return early_exit(REASON_ADB_SERVER_FAILED, s.message) + + with events.step("list_devices") as s: + listed_devices = AdbWrapper.devices(logger) + if listed_devices is None: + s.fail("`adb devices` failed.") + connected_devices = listed_devices or [] # With --serial, fall through: every requested serial is then reported as # missing, FAILED in the summary, and the run exits 1. if not connected_devices and args.serial is None: logger.error("No devices connected!") - return + if listed_devices is None: + return early_exit(REASON_ADB_DEVICES_FAILED, "`adb devices` failed.") + return early_exit(REASON_NO_DEVICES, "No devices connected!") logger.debug("There are %d connected device(s)", len(connected_devices)) target_devices, missing_serials = select_target_devices( @@ -963,76 +1281,108 @@ def main(): if args.serial is None and len(connected_devices) > 1: logger.warning("More than 1 device connected!") + events.emit("devices", + devices=[_device_to_event(d) for d in connected_devices], + selected=[d.serial_number for d in target_devices], + missing=missing_serials) + for serial in missing_serials: + events.device_finished(serial, STATUS_FAILED, + error=_error(REASON_NOT_CONNECTED, + "Requested device is not connected.")) + results = {} prefetched = False has_errors = bool(missing_serials) verification_failed_devices = set() collection_error_devices = set() for target_device in target_devices: + serial = target_device.serial_number + device_error = None + events.emit("device_started", device=serial) try: if target_device.unauthorized: logger.error("Please authorize device with serial number %s for ADB via " - "device GUI.", target_device.serial_number) + "device GUI.", serial) + device_error = _error(REASON_UNAUTHORIZED, + "ADB is not authorized on this device.") has_errors = True continue # set up an adb_wrapper to be used throughout for this target device - adb_wrapper = AdbWrapper(target_device.serial_number, logger) + adb_wrapper = AdbWrapper(serial, logger) if is_hubble_installed(adb_wrapper, logger): logger.debug("Removing previous Hubble installation...") - if not remove_previous_installation(adb_wrapper): - logger.error("Failed to remove previous Hubble installation.") + with events.step("uninstall_previous", serial) as s: + if not remove_previous_installation(adb_wrapper): + logger.error("Failed to remove previous Hubble installation.") + device_error = _error(REASON_UNINSTALL_FAILED, s.fail( + "Failed to remove previous Hubble installation.")) + has_errors = True + continue + + with events.step("install_hubble", serial) as s: + if is_xiaomi_phone(adb_wrapper, logger): + logger.info("This is a Xiaomi phone.") + adb_push_hubble(adb_wrapper, args.hubble) + if not launch_xiaomi_file_explorer(adb_wrapper): + logger.error("Failed to launch Xiaomi file explorer") + while not is_hubble_installed(adb_wrapper, logger): + logger.warning("Please manually install Hubble by launching the " + "\"Files Manager\" app (it may have been launched " + "for you) and navigate to the \"Downloads\" folder.") + input("Press [ENTER] when you are done.") + else: + logger.info("This is not a Xiaomi phone. Regular workflow continues...") + if not install_hubble(adb_wrapper, args, logger): + logger.error("Error installing Hubble: %s", adb_wrapper.error_message) + device_error = _error(REASON_INSTALL_FAILED, s.fail( + "Error installing Hubble: {}".format( + adb_wrapper.error_message))) + has_errors = True + continue + + with events.step("launch_hubble", serial) as s: + clear_logcat(adb_wrapper) + if not launch_hubble(adb_wrapper): + logger.error("Failed to launch Hubble: %s", adb_wrapper.error_message) + device_error = _error(REASON_LAUNCH_FAILED, s.fail( + "Failed to launch Hubble: {}".format(adb_wrapper.error_message))) has_errors = True continue - if is_xiaomi_phone(adb_wrapper, logger): - logger.info("This is a Xiaomi phone.") - adb_push_hubble(adb_wrapper, args.hubble) - if not launch_xiaomi_file_explorer(adb_wrapper): - logger.error("Failed to launch Xiaomi file explorer") - while not is_hubble_installed(adb_wrapper, logger): - logger.warning("Please manually install Hubble by launching the " - "\"Files Manager\" app (it may have been launched " - "for you) and navigate to the \"Downloads\" folder.") - input("Press [ENTER] when you are done.") - else: - logger.info("This is not a Xiaomi phone. Regular workflow continues...") - if not install_hubble(adb_wrapper, args, logger): - logger.error("Error installing Hubble: %s", adb_wrapper.error_message) + with events.step("wait_for_results", serial) as s: + results_source = wait_for_results(adb_wrapper, logger) + if not results_source: + logger.error("Failed to obtain results from Hubble execution.") + device_error = _error(REASON_NO_RESULTS, s.fail( + "Failed to obtain results from Hubble execution.")) has_errors = True continue - clear_logcat(adb_wrapper) - if not launch_hubble(adb_wrapper): - logger.error("Failed to launch Hubble: %s", adb_wrapper.error_message) - has_errors = True - continue - results_source = wait_for_results(adb_wrapper, logger) - - if not results_source: - logger.error("Failed to obtain results from Hubble execution.") - has_errors = True - continue - extract_apks = (args.pull_all_apks is not None) or args.pull_preinstalled_apks_only - with tempfile.TemporaryDirectory() as device_tmp_dir: - results_dir = extract_results_and_apks( - adb_wrapper, - results_source, - args.output, - logger, - extract_apks, - pull_preinstalled_only=args.pull_preinstalled_apks_only, - tmp_dir=device_tmp_dir) - - if not results_dir: - logger.error("Failed to extract results from target device (%s).", - target_device.serial_number) - has_errors = True - continue + with events.step("extract_results", serial) as s: + extract_apks = (args.pull_all_apks is not None) or args.pull_preinstalled_apks_only + with tempfile.TemporaryDirectory() as device_tmp_dir: + results_dir = extract_results_and_apks( + adb_wrapper, + results_source, + args.output, + logger, + extract_apks, + pull_preinstalled_only=args.pull_preinstalled_apks_only, + tmp_dir=device_tmp_dir) + + if not results_dir: + logger.error("Failed to extract results from target device (%s).", + serial) + device_error = _error(REASON_EXTRACT_FAILED, s.fail( + "Failed to extract results from device.")) + has_errors = True + continue - extract_selinux_policies(adb_wrapper, results_dir, logger) - results[target_device.serial_number] = results_dir + with events.step("extract_selinux", serial): + extract_selinux_policies(adb_wrapper, results_dir, logger) + results[serial] = results_dir if args.perform_inclusion_proof_check: pkg_filename = ( @@ -1042,38 +1392,73 @@ def main(): ) packages_txt_path = os.path.join(results_dir, "results", pkg_filename) if args.check_preinstalled_only and not os.path.isfile(packages_txt_path): - logger.error( - "preinstalled_packages.txt not found at %s (extracted results at %s). " - "--check_preinstalled_only requires Hubble >= 2.1.0.", - packages_txt_path, results_dir) + with events.step("inclusion_proof_check", serial) as s: + logger.error( + "preinstalled_packages.txt not found at %s (extracted results at %s). " + "--check_preinstalled_only requires Hubble >= 2.1.0.", + packages_txt_path, results_dir) + device_error = _error( + REASON_INCLUSION_PROOF_CHECK_INCOMPLETE, + s.fail("preinstalled_packages.txt not found; " + "--check_preinstalled_only requires Hubble >= 2.1.0.")) has_errors = True - verification_failed_devices.add(target_device.serial_number) + verification_failed_devices.add(serial) continue if (not args.no_prefetch and not prefetched and os.path.isfile(packages_txt_path)): - prefetched = inclusion_proof_check.prefetch_log_entries( + with events.step("inclusion_proof_prefetch", serial) as s: + prefetched = inclusion_proof_check.prefetch_log_entries( + args.verifier_path, + logger, + cache_dir=args.cache_dir, + concurrency=args.cache_prefetch_concurrency, + timeout=args.cache_prefetch_timeout) + if not prefetched: + s.fail("Pre-fetching failed; falling back to on-demand " + "fetching.") + with events.step("inclusion_proof_check", serial) as s: + if not inclusion_proof_check.perform_inclusion_proof_check( args.verifier_path, + packages_txt_path, logger, cache_dir=args.cache_dir, concurrency=args.cache_prefetch_concurrency, - timeout=args.cache_prefetch_timeout) - if not inclusion_proof_check.perform_inclusion_proof_check( - args.verifier_path, - packages_txt_path, - logger, - cache_dir=args.cache_dir, - concurrency=args.cache_prefetch_concurrency, - timeout=args.cache_prefetch_timeout, - prefetch=False, - preinstalled_only=args.check_preinstalled_only): - has_errors = True - verification_failed_devices.add(target_device.serial_number) + timeout=args.cache_prefetch_timeout, + prefetch=False, + preinstalled_only=args.check_preinstalled_only): + has_errors = True + verification_failed_devices.add(serial) + # False means the check could not complete (bad input or + # unwritable output), not that some splits are absent from the + # log; per-split results are in the *_signal.txt output. + device_error = _error(REASON_INCLUSION_PROOF_CHECK_INCOMPLETE, + s.fail("Inclusion proof check could not " + "complete.")) except Exception as e: logger.exception("Unexpected error while processing device %s: %s", - target_device.serial_number, e) + serial, e) + device_error = _error(REASON_UNEXPECTED_ERROR, + "{}: {}".format(type(e).__name__, e)) has_errors = True - if target_device.serial_number in results: - collection_error_devices.add(target_device.serial_number) + if serial in results: + collection_error_devices.add(serial) + except BaseException as e: + # KeyboardInterrupt and other non-Exception exits bypass the handler + # above. Record them before propagating, so the device_finished event + # in the finally block does not report an interrupted device as a + # success when its results were already collected. + device_error = _interruption_error(e) + if serial in results: + collection_error_devices.add(serial) + raise + finally: + events.device_finished( + serial, + device_status(serial, results, missing_serials, + collection_error_devices, + verification_failed_devices), + results_dir=results.get(serial), + error=device_error) # Summarise in the order devices were requested (or discovered, without # --serial), including requested serials that were never connected. @@ -1081,25 +1466,33 @@ def main(): summary_serials = list(dict.fromkeys( args.serial if args.serial is not None else [d.serial_number for d in connected_devices])) + summary = {} for device in summary_serials: + status = device_status(device, results, missing, + collection_error_devices, + verification_failed_devices) + summary[device] = {"status": status} + if device in results: + summary[device]["results_dir"] = results[device] + if device in missing: logger.error( "FAILED: Requested device %s is not connected (exiting 1).", device) continue - if device not in results: + if status == STATUS_FAILED: logger.error( "FAILED: Hubble data collection failed on connected device %s " "(exiting 1).", device) continue - if device in collection_error_devices: + if status == STATUS_PARTIAL_ERROR: logger.warning( "PARTIAL SUCCESS: Hubble data collection succeeded on connected " "device %s, but an unexpected error occurred during post-collection " "processing (exiting 1).", device) - elif device in verification_failed_devices: + elif status == STATUS_PARTIAL_CHECK_INCOMPLETE: logger.warning( "PARTIAL SUCCESS: Hubble data collection succeeded on connected " "device %s, but inclusion proof verification failed (exiting 1).", @@ -1109,8 +1502,66 @@ def main(): "connected device %s.", device) logger.info("Hubble output files can be found at: %s", results[device]) - if has_errors: - sys.exit(1) + exit_code = 1 if has_errors else 0 + events.finish_run(exit_code, summary=summary) + return exit_code + + +def main(): + args = parse_arguments() + logger = set_up_logging(args) + + events = EventEmitter(logger=logger) + if args.events is not None: + try: + events = EventEmitter(open_event_stream(args.events), logger=logger) + except OSError as e: + logger.error("Cannot open --events destination %s: %s", args.events, e) + sys.exit(1) + + # SIGTERM (how CI timeouts and cancellations usually stop a process) would + # otherwise kill the script without a run_finished event. While events are + # being written, turn it into an exception so the run can report itself, + # then die by SIGTERM anyway so the parent sees the usual exit status. + # Without --events, SIGTERM handling is left untouched. + previous_sigterm_handler = None + if events.enabled: + previous_sigterm_handler = signal.signal(signal.SIGTERM, _raise_terminated) + + terminated = False + try: + exit_code = run(args, logger, events) + except Terminated: + events.finish_run(128 + signal.SIGTERM, error=_error( + REASON_TERMINATED, "Terminated by SIGTERM.")) + terminated = True + except KeyboardInterrupt: + events.finish_run(130, error=_error(REASON_INTERRUPTED, + "Interrupted by user.")) + raise + except BaseException as e: + events.finish_run(1, error=_error( + REASON_UNEXPECTED_ERROR, "{}: {}".format(type(e).__name__, e))) + raise + finally: + events.close() + if previous_sigterm_handler is not None: + signal.signal(signal.SIGTERM, previous_sigterm_handler) + + if terminated: + _die_by_sigterm(logger) + return + if exit_code: + sys.exit(exit_code) + + +def _die_by_sigterm(logger): + """Terminates this process with SIGTERM's default action.""" + logger.error("Terminated by SIGTERM.") + signal.signal(signal.SIGTERM, signal.SIG_DFL) + os.kill(os.getpid(), signal.SIGTERM) + # Not reached unless SIGTERM is blocked; fall back to the conventional code. + sys.exit(128 + signal.SIGTERM) if __name__ == "__main__": diff --git a/uraniborg/scripts/python/tests/test_automate_observation_events.py b/uraniborg/scripts/python/tests/test_automate_observation_events.py new file mode 100644 index 0000000000..aa7e2af --- /dev/null +++ b/uraniborg/scripts/python/tests/test_automate_observation_events.py @@ -0,0 +1,879 @@ +#!/usr/bin/python3 +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +"""Unit tests for automate_observation.py's --events progress stream.""" + +import contextlib +import datetime +import io +import json +import os +from pathlib import Path +import signal +import subprocess +import sys +import textwrap +import time +from unittest import mock + +import pytest + +# Ensure uraniborg/scripts/python is on sys.path +SCRIPT_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) +sys.path.insert(0, SCRIPT_DIR) + +import automate_observation +# Shared fixture and helpers for driving main() with every collaborator mocked. +from test_automate_observation import _make_mock_device # pylint: disable=g-importing-member +from test_automate_observation import _set_argv # pylint: disable=g-importing-member +from test_automate_observation import serial_main_mocks # pylint: disable=unused-import,g-importing-member + +EventEmitter = automate_observation.EventEmitter + +_FIXED_TIME = datetime.datetime(2026, 1, 2, 3, 4, 5, 678000, + tzinfo=datetime.timezone.utc) + + +def _emitter(stream=None, logger=None): + return EventEmitter(stream if stream is not None else io.StringIO(), + logger=logger, clock=lambda: _FIXED_TIME) + + +def _parse(stream: io.StringIO) -> list[dict]: + return [json.loads(line) for line in stream.getvalue().splitlines()] + + +def _read_events(path: Path) -> list[dict]: + return [json.loads(line) for line in path.read_text().splitlines()] + + +def _steps(events: list[dict], device=None) -> list[tuple[str, str]]: + return [(e["step"], e["state"]) for e in events + if e["type"] == "step" and e.get("device") == device] + + +def _of_type(events: list[dict], event_type: str) -> list[dict]: + return [e for e in events if e["type"] == event_type] + + +# --- EventEmitter --------------------------------------------------------------- + + +def test_emitter_without_stream_is_a_noop(): + emitter = EventEmitter() + assert not emitter.enabled + emitter.emit("anything", x=1) + with emitter.step("s") as s: + s.fail("nope") + emitter.finish_run(1) + emitter.close() # must not raise + + +def test_emit_envelope_drops_none_and_flushes_each_event(): + stream = mock.Mock(wraps=io.StringIO()) + emitter = EventEmitter(stream, clock=lambda: _FIXED_TIME) + + emitter.emit("thing", device="D1", results_dir=None, count=2) + assert stream.flush.call_count == 1 + emitter.emit("other") + assert stream.flush.call_count == 2 + + lines = stream.getvalue().splitlines() + assert json.loads(lines[0]) == { + "v": 1, "ts": "2026-01-02T03:04:05.678Z", "type": "thing", + "device": "D1", "count": 2, + } + assert json.loads(lines[1]) == { + "v": 1, "ts": "2026-01-02T03:04:05.678Z", "type": "other"} + + +def test_emit_write_failure_warns_once_and_disables(): + stream = mock.Mock() + stream.write.side_effect = BrokenPipeError("reader went away") + logger = mock.Mock() + emitter = EventEmitter(stream, logger=logger) + + emitter.emit("a") + emitter.emit("b") + + assert not emitter.enabled + assert stream.write.call_count == 1 + logger.warning.assert_called_once() + + +def test_step_reports_finished_failed_and_exceptions(): + stream = io.StringIO() + emitter = _emitter(stream) + + with emitter.step("ok", device="D1"): + pass + with emitter.step("soft_fail") as s: + assert s.fail("went wrong") == "went wrong" + with pytest.raises(RuntimeError): + with emitter.step("boom", device="D1"): + raise RuntimeError("kaput") + + events = _parse(stream) + assert [(e["step"], e["state"], e.get("device")) for e in events] == [ + ("ok", "started", "D1"), ("ok", "finished", "D1"), + ("soft_fail", "started", None), ("soft_fail", "failed", None), + ("boom", "started", "D1"), ("boom", "failed", "D1"), + ] + assert "message" not in events[1] + assert events[3]["message"] == "went wrong" + assert events[5]["message"] == "RuntimeError: kaput" + for e in events[1::2]: + assert isinstance(e["duration_ms"], int) and e["duration_ms"] >= 0 + for e in events[0::2]: + assert "duration_ms" not in e + + +def test_step_reports_finished_when_body_continues_a_loop(): + """`continue` inside the with-block must still close the step.""" + stream = io.StringIO() + emitter = _emitter(stream) + for _ in range(1): + with emitter.step("loop") as s: + s.fail("skipped") + continue + assert [(e["step"], e["state"]) for e in _parse(stream)] == [ + ("loop", "started"), ("loop", "failed")] + + +def test_finish_run_is_emitted_once_with_ok_semantics(): + stream = io.StringIO() + emitter = _emitter(stream) + emitter.finish_run(0, summary={"D1": {"status": "success"}}) + emitter.finish_run(1, error={"reason": "x", "message": "y"}) # ignored + (event,) = _parse(stream) + assert event["type"] == "run_finished" + assert event["exit_code"] == 0 + assert event["ok"] is True + assert event["summary"] == {"D1": {"status": "success"}} + assert "error" not in event + + for exit_code, error, ok in [(1, None, False), + (0, {"reason": "r", "message": "m"}, False)]: + stream = io.StringIO() + _emitter(stream).finish_run(exit_code, error=error) + (event,) = _parse(stream) + assert event["ok"] is ok + assert event["summary"] == {} + + +@pytest.mark.parametrize( + "serial, expected", + [ + ("MISSING", automate_observation.STATUS_FAILED), + ("NO_RESULTS", automate_observation.STATUS_FAILED), + ("POST_ERR", automate_observation.STATUS_PARTIAL_ERROR), + ("BOTH", automate_observation.STATUS_PARTIAL_ERROR), + ("VERIFY", automate_observation.STATUS_PARTIAL_CHECK_INCOMPLETE), + ("OK", automate_observation.STATUS_SUCCESS), + ], +) +def test_device_status(serial, expected): + results = {s: "/r/" + s for s in ("POST_ERR", "BOTH", "VERIFY", "OK")} + assert automate_observation.device_status( + serial, results, {"MISSING"}, {"POST_ERR", "BOTH"}, + {"VERIFY", "BOTH"}) == expected + + +def test_device_to_event_skips_empty_and_non_string_attributes(): + dev = automate_observation.syscall_wrapper.DeviceInfo() + dev.serial_number = "S1" + dev.model_name = "Pixel_9" + dev.product_name = "" + assert automate_observation._device_to_event(dev) == { + "serial": "S1", "unauthorized": False, "model": "Pixel_9"} + # Mock devices (as used throughout the tests) must not leak Mock reprs. + assert automate_observation._device_to_event(_make_mock_device("S2")) == { + "serial": "S2", "unauthorized": False} + + +# --- main() with --events -------------------------------------------------------- + + +def test_main_events_full_run_with_missing_serial( + serial_main_mocks, monkeypatch: pytest.MonkeyPatch, tmp_path: Path, +): + m = serial_main_mocks + unauthorized = _make_mock_device("DEV_UNAUTH") + unauthorized.unauthorized = True + m["AdbWrapper"].devices.return_value = [ + _make_mock_device("DEV1"), unauthorized, _make_mock_device("DEV_OTHER")] + events_path = tmp_path / "events.jsonl" + _set_argv(monkeypatch, "--events", str(events_path), + "-s", "DEV1", "-s", "GONE", "-s", "DEV_UNAUTH") + + with pytest.raises(SystemExit) as exc_info: + automate_observation.main() + assert exc_info.value.code == 1 + + events = _read_events(events_path) + assert all(e["v"] == 1 and e["ts"].endswith("Z") for e in events) + assert events[0]["type"] == "run_started" + assert events[0]["argv"][-6:] == [ + "-s", "DEV1", "-s", "GONE", "-s", "DEV_UNAUTH"] + assert events[-1]["type"] == "run_finished" + assert len(_of_type(events, "run_finished")) == 1 + + assert _steps(events) == [ + ("verify_hubble", "started"), ("verify_hubble", "finished"), + ("check_adb", "started"), ("check_adb", "finished"), + ("start_adb_server", "started"), ("start_adb_server", "finished"), + ("list_devices", "started"), ("list_devices", "finished"), + ] + + (devices,) = _of_type(events, "devices") + assert devices["devices"] == [ + {"serial": "DEV1", "unauthorized": False}, + {"serial": "DEV_UNAUTH", "unauthorized": True}, + {"serial": "DEV_OTHER", "unauthorized": False}, + ] + assert devices["selected"] == ["DEV1", "DEV_UNAUTH"] + assert devices["missing"] == ["GONE"] + + # Every selected or missing serial gets exactly one device_finished; + # missing ones get no device_started. + assert [e["device"] for e in _of_type(events, "device_started")] == [ + "DEV1", "DEV_UNAUTH"] + finished = {e["device"]: e for e in _of_type(events, "device_finished")} + assert set(finished) == {"GONE", "DEV1", "DEV_UNAUTH"} + assert finished["GONE"]["status"] == "failed" + assert finished["GONE"]["error"] == { + "reason": "not_connected", + "message": "Requested device is not connected."} + assert finished["DEV1"]["status"] == "success" + assert finished["DEV1"]["results_dir"] == "/tmp/out/DEV1" + assert "error" not in finished["DEV1"] + assert finished["DEV_UNAUTH"]["status"] == "failed" + assert "results_dir" not in finished["DEV_UNAUTH"] + assert finished["DEV_UNAUTH"]["error"]["reason"] == "unauthorized" + assert "not authorized" in finished["DEV_UNAUTH"]["error"]["message"] + + assert _steps(events, "DEV1") == [ + ("install_hubble", "started"), ("install_hubble", "finished"), + ("launch_hubble", "started"), ("launch_hubble", "finished"), + ("wait_for_results", "started"), ("wait_for_results", "finished"), + ("extract_results", "started"), ("extract_results", "finished"), + ("extract_selinux", "started"), ("extract_selinux", "finished"), + ] + assert _steps(events, "DEV_UNAUTH") == [] + + run_finished = events[-1] + assert run_finished["exit_code"] == 1 + assert run_finished["ok"] is False + assert "error" not in run_finished + assert run_finished["summary"] == { + "DEV1": {"status": "success", "results_dir": "/tmp/out/DEV1"}, + "GONE": {"status": "failed"}, + "DEV_UNAUTH": {"status": "failed"}, + } + assert list(run_finished["summary"]) == ["DEV1", "GONE", "DEV_UNAUTH"] + + +def test_main_events_success_exits_0_with_ok( + serial_main_mocks, monkeypatch: pytest.MonkeyPatch, tmp_path: Path, +): + m = serial_main_mocks + m["AdbWrapper"].devices.return_value = [_make_mock_device("DEV1")] + events_path = tmp_path / "events.jsonl" + _set_argv(monkeypatch, "--events", str(events_path)) + + automate_observation.main() # no SystemExit + + run_finished = _read_events(events_path)[-1] + assert run_finished["type"] == "run_finished" + assert run_finished["exit_code"] == 0 + assert run_finished["ok"] is True + + +def test_main_events_previous_install_and_install_failure( + serial_main_mocks, monkeypatch: pytest.MonkeyPatch, tmp_path: Path, +): + m = serial_main_mocks + m["AdbWrapper"].devices.return_value = [_make_mock_device("DEV1")] + m["is_hubble_installed"].return_value = True + with mock.patch("automate_observation.remove_previous_installation", + return_value=True): + m["install_hubble"].return_value = False + events_path = tmp_path / "events.jsonl" + _set_argv(monkeypatch, "--events", str(events_path)) + with pytest.raises(SystemExit): + automate_observation.main() + + events = _read_events(events_path) + install_failed = [e for e in events if e["type"] == "step" + and e["step"] == "install_hubble" + and e["state"] == "failed"] + assert _steps(events, "DEV1") == [ + ("uninstall_previous", "started"), ("uninstall_previous", "finished"), + ("install_hubble", "started"), ("install_hubble", "failed"), + ] + assert install_failed[0]["message"].startswith("Error installing Hubble:") + (finished,) = _of_type(events, "device_finished") + assert finished["status"] == "failed" + assert finished["error"] == {"reason": "install_failed", + "message": install_failed[0]["message"]} + + +@pytest.mark.parametrize( + "prefetch_ok, check_result, expected_status, expected_check_state", + [ + (True, True, "success", "finished"), + (False, True, "success", "finished"), + (True, False, "partial_check_incomplete", "failed"), + (True, RuntimeError("verifier crashed"), "partial_error", "failed"), + ], + ids=["verified", "prefetch_failed", "verification_failed", "crash"], +) +def test_main_events_inclusion_proof_outcomes( + serial_main_mocks, monkeypatch: pytest.MonkeyPatch, tmp_path: Path, + prefetch_ok, check_result, expected_status, expected_check_state, +): + m = serial_main_mocks + m["AdbWrapper"].devices.return_value = [_make_mock_device("DEV1")] + events_path = tmp_path / "events.jsonl" + _set_argv(monkeypatch, "--events", str(events_path), + "--perform_inclusion_proof_check", "--verifier_path=/v") + check_kwargs = ({"side_effect": check_result} + if isinstance(check_result, Exception) + else {"return_value": check_result}) + with mock.patch("automate_observation.os.path.isfile", return_value=True), \ + mock.patch("inclusion_proof_check.prefetch_log_entries", + return_value=prefetch_ok), \ + mock.patch("inclusion_proof_check.perform_inclusion_proof_check", + **check_kwargs): + if expected_status == "success": + automate_observation.main() + else: + with pytest.raises(SystemExit): + automate_observation.main() + + events = _read_events(events_path) + steps = _steps(events, "DEV1") + assert steps[-4:] == [ + ("inclusion_proof_prefetch", "started"), + ("inclusion_proof_prefetch", "finished" if prefetch_ok else "failed"), + ("inclusion_proof_check", "started"), + ("inclusion_proof_check", expected_check_state), + ] + (finished,) = _of_type(events, "device_finished") + assert finished["status"] == expected_status + # Results were collected in every case, so the directory is always reported. + assert finished["results_dir"] == "/tmp/out/DEV1" + assert events[-1]["summary"]["DEV1"]["status"] == expected_status + if expected_status == "partial_error": + assert finished["error"] == {"reason": "unexpected_error", + "message": "RuntimeError: verifier crashed"} + if expected_status == "partial_check_incomplete": + # A False return means the check could not complete, not that APKs are + # missing from the log; the message must not claim otherwise. + assert finished["error"] == { + "reason": "inclusion_proof_check_incomplete", + "message": "Inclusion proof check could not complete."} + + +def test_main_events_check_preinstalled_only_missing_file( + serial_main_mocks, monkeypatch: pytest.MonkeyPatch, tmp_path: Path, +): + m = serial_main_mocks + m["AdbWrapper"].devices.return_value = [_make_mock_device("DEV1")] + events_path = tmp_path / "events.jsonl" + _set_argv(monkeypatch, "--events", str(events_path), + "--perform_inclusion_proof_check", "--verifier_path=/v", + "--check_preinstalled_only") + with mock.patch("automate_observation.os.path.isfile", return_value=False): + with pytest.raises(SystemExit): + automate_observation.main() + + events = _read_events(events_path) + assert _steps(events, "DEV1")[-2:] == [ + ("inclusion_proof_check", "started"), + ("inclusion_proof_check", "failed"), + ] + (finished,) = _of_type(events, "device_finished") + assert finished["status"] == "partial_check_incomplete" + assert finished["error"]["reason"] == "inclusion_proof_check_incomplete" + assert "preinstalled_packages.txt not found" in finished["error"]["message"] + + +def _fail_uninstall(m, stack): + m["is_hubble_installed"].return_value = True + stack.enter_context(mock.patch( + "automate_observation.remove_previous_installation", return_value=False)) + + +def _fail_install(m, stack): # pylint: disable=unused-argument + m["install_hubble"].return_value = False + + +def _fail_launch(m, stack): # pylint: disable=unused-argument + m["launch_hubble"].return_value = False + + +def _fail_wait(m, stack): # pylint: disable=unused-argument + m["wait_for_results"].return_value = None + + +def _fail_extract(m, stack): # pylint: disable=unused-argument + m["extract_results_and_apks"].side_effect = lambda *a, **kw: None + + +@pytest.mark.parametrize( + "setup, failed_step, reason", + [ + (_fail_uninstall, "uninstall_previous", "uninstall_failed"), + (_fail_install, "install_hubble", "install_failed"), + (_fail_launch, "launch_hubble", "launch_failed"), + (_fail_wait, "wait_for_results", "no_results"), + (_fail_extract, "extract_results", "extract_failed"), + ], + ids=["uninstall", "install", "launch", "wait", "extract"], +) +def test_main_events_device_error_reason_per_failed_step( + serial_main_mocks, monkeypatch: pytest.MonkeyPatch, tmp_path: Path, + setup, failed_step, reason, +): + m = serial_main_mocks + m["AdbWrapper"].devices.return_value = [_make_mock_device("DEV1")] + events_path = tmp_path / "events.jsonl" + _set_argv(monkeypatch, "--events", str(events_path)) + with contextlib.ExitStack() as stack: + setup(m, stack) + with pytest.raises(SystemExit): + automate_observation.main() + + events = _read_events(events_path) + (step_failed,) = [e for e in events if e["type"] == "step" + and e["state"] == "failed"] + assert step_failed["step"] == failed_step + (finished,) = _of_type(events, "device_finished") + assert finished["status"] == "failed" + # The reason is a stable code; the message matches the failed step's. + assert finished["error"] == {"reason": reason, + "message": step_failed["message"]} + + +@pytest.mark.parametrize( + "mock_name, value, reason, failed_step", + [ + ("supported_platform", False, "unsupported_platform", None), + ("verify_hubble", False, "invalid_hubble_apk", "verify_hubble"), + ("adb_installed", False, "adb_not_found", "check_adb"), + ("AdbWrapper.start_server", False, "adb_server_failed", + "start_adb_server"), + ("AdbWrapper.devices", [], "no_devices", None), + ("AdbWrapper.devices", None, "adb_devices_failed", "list_devices"), + ], +) +def test_main_events_early_exits_keep_exit_0_and_report_reason( + serial_main_mocks, monkeypatch: pytest.MonkeyPatch, tmp_path: Path, + mock_name, value, reason, failed_step, +): + m = serial_main_mocks + if "." in mock_name: + owner, attr = mock_name.split(".") + getattr(m[owner], attr).return_value = value + else: + m[mock_name].return_value = value + events_path = tmp_path / "events.jsonl" + _set_argv(monkeypatch, "--events", str(events_path)) + + automate_observation.main() # early exits still return exit code 0 + + events = _read_events(events_path) + assert events[0]["type"] == "run_started" + run_finished = events[-1] + assert run_finished["type"] == "run_finished" + assert len(_of_type(events, "run_finished")) == 1 + assert run_finished["exit_code"] == 0 + assert run_finished["ok"] is False + assert run_finished["summary"] == {} + assert run_finished["error"]["reason"] == reason + assert run_finished["error"]["message"] + failed = [e["step"] for e in events + if e["type"] == "step" and e["state"] == "failed"] + assert failed == ([failed_step] if failed_step else []) + assert _of_type(events, "devices") == [] + + +def test_main_events_build_failure( + serial_main_mocks, monkeypatch: pytest.MonkeyPatch, tmp_path: Path, +): + events_path = tmp_path / "events.jsonl" + monkeypatch.setattr(sys, "argv", ["automate_observation.py", + "--events", str(events_path)]) + with mock.patch("automate_observation.ensure_android_sdk", + return_value=False): + automate_observation.main() + + events = _read_events(events_path) + assert _steps(events) == [("build_hubble", "started"), + ("build_hubble", "failed")] + assert events[-1]["error"] == { + "reason": "android_sdk_not_found", + "message": "Failed to (re)build Hubble APK: Android SDK not found.", + } + serial_main_mocks["verify_hubble"].assert_not_called() + + +def test_main_events_unexpected_exception_reports_and_reraises( + serial_main_mocks, monkeypatch: pytest.MonkeyPatch, tmp_path: Path, +): + m = serial_main_mocks + m["AdbWrapper"].devices.side_effect = ValueError("bad adb output") + events_path = tmp_path / "events.jsonl" + _set_argv(monkeypatch, "--events", str(events_path)) + + with pytest.raises(ValueError): + automate_observation.main() + + events = _read_events(events_path) + assert _steps(events)[-1] == ("list_devices", "failed") + assert events[-1]["type"] == "run_finished" + assert events[-1]["exit_code"] == 1 + assert events[-1]["error"] == {"reason": "unexpected_error", + "message": "ValueError: bad adb output"} + + +@pytest.mark.parametrize( + "interrupt_at, expected_status", + [ + ("wait_for_results", "failed"), + # After results are collected: must not be reported as success. + ("prefetch", "partial_error"), + ("check", "partial_error"), + ], +) +def test_main_events_keyboard_interrupt_during_device( + serial_main_mocks, monkeypatch: pytest.MonkeyPatch, tmp_path: Path, + interrupt_at, expected_status, +): + m = serial_main_mocks + m["AdbWrapper"].devices.return_value = [_make_mock_device("DEV1")] + events_path = tmp_path / "events.jsonl" + _set_argv(monkeypatch, "--events", str(events_path), + "--perform_inclusion_proof_check", "--verifier_path=/v") + if interrupt_at == "wait_for_results": + m["wait_for_results"].side_effect = KeyboardInterrupt + prefetch = mock.patch( + "inclusion_proof_check.prefetch_log_entries", + side_effect=KeyboardInterrupt if interrupt_at == "prefetch" else None, + return_value=True) + check = mock.patch( + "inclusion_proof_check.perform_inclusion_proof_check", + side_effect=KeyboardInterrupt if interrupt_at == "check" else None, + return_value=True) + + with mock.patch("automate_observation.os.path.isfile", return_value=True), \ + prefetch, check: + with pytest.raises(KeyboardInterrupt): + automate_observation.main() + + events = _read_events(events_path) + (finished,) = _of_type(events, "device_finished") + assert finished["status"] == expected_status + assert finished["error"] == {"reason": "interrupted", + "message": "Interrupted."} + if expected_status == "partial_error": + assert finished["results_dir"] == "/tmp/out/DEV1" + assert events[-1]["type"] == "run_finished" + assert events[-1]["exit_code"] == 130 + assert events[-1]["error"]["reason"] == "interrupted" + + +def _two_devices_interrupted_on_second(m, exc): + """DEV1 completes; DEV2 raises exc while waiting for Hubble's results.""" + m["AdbWrapper"].devices.return_value = [ + _make_mock_device("DEV1"), _make_mock_device("DEV2")] + m["wait_for_results"].side_effect = [ + "/sdcard/hubble/results", exc] + + +def test_main_events_interrupt_reports_partial_summary( + serial_main_mocks, monkeypatch: pytest.MonkeyPatch, tmp_path: Path, +): + """Devices that finished before Ctrl-C stay in run_finished.summary.""" + m = serial_main_mocks + _two_devices_interrupted_on_second(m, KeyboardInterrupt) + events_path = tmp_path / "events.jsonl" + _set_argv(monkeypatch, "--events", str(events_path), + "-s", "GONE", "-s", "DEV1", "-s", "DEV2") + + with pytest.raises(KeyboardInterrupt): + automate_observation.main() + + run_finished = _read_events(events_path)[-1] + assert run_finished["error"]["reason"] == "interrupted" + assert run_finished["summary"] == { + "GONE": {"status": "failed"}, + "DEV1": {"status": "success", "results_dir": "/tmp/out/DEV1"}, + "DEV2": {"status": "failed"}, + } + + +def test_main_events_unexpected_error_after_devices_reports_partial_summary( + serial_main_mocks, monkeypatch: pytest.MonkeyPatch, tmp_path: Path, +): + m = serial_main_mocks + m["AdbWrapper"].devices.return_value = [_make_mock_device("DEV1")] + + def _info(msg, *args): + if msg.startswith("SUCCESS!"): # fail while printing the log summary + raise RuntimeError("logging broke") + m["logger"].info.side_effect = _info + events_path = tmp_path / "events.jsonl" + _set_argv(monkeypatch, "--events", str(events_path)) + + with pytest.raises(RuntimeError): + automate_observation.main() + + run_finished = _read_events(events_path)[-1] + assert run_finished["error"]["reason"] == "unexpected_error" + assert run_finished["summary"] == { + "DEV1": {"status": "success", "results_dir": "/tmp/out/DEV1"}} + + +def test_finish_run_defaults_to_recorded_devices_and_explicit_summary_wins(): + stream = io.StringIO() + emitter = _emitter(stream) + emitter.device_finished("A", "success", results_dir="/r/A") + emitter.device_finished("B", "failed", + error={"reason": "unexpected_error", + "message": "boom"}) + emitter.finish_run(1) + events = _parse(stream) + assert events[0] == {"v": 1, "ts": "2026-01-02T03:04:05.678Z", + "type": "device_finished", "device": "A", + "status": "success", "results_dir": "/r/A"} + assert events[1]["error"] == {"reason": "unexpected_error", + "message": "boom"} + assert events[-1]["summary"] == {"A": {"status": "success", + "results_dir": "/r/A"}, + "B": {"status": "failed"}} + + stream = io.StringIO() + emitter = _emitter(stream) + emitter.device_finished("A", "success") + emitter.finish_run(0, summary={"X": {"status": "success"}}) + assert _parse(stream)[-1]["summary"] == {"X": {"status": "success"}} + + +def test_main_sigterm_handler_only_installed_with_events( + serial_main_mocks, monkeypatch: pytest.MonkeyPatch, tmp_path: Path, +): + """Without --events SIGTERM is untouched; with it, restored afterwards.""" + m = serial_main_mocks + m["AdbWrapper"].devices.return_value = [_make_mock_device("DEV1")] + seen = [] + + def _wait(*unused_args): + seen.append(signal.getsignal(signal.SIGTERM)) + return "/sdcard/hubble/results" + m["wait_for_results"].side_effect = _wait + before = signal.getsignal(signal.SIGTERM) + + _set_argv(monkeypatch) + automate_observation.main() + _set_argv(monkeypatch, "--events", str(tmp_path / "events.jsonl")) + automate_observation.main() + + assert seen == [before, automate_observation._raise_terminated] + assert signal.getsignal(signal.SIGTERM) == before + + +def test_main_events_terminated_in_process( + serial_main_mocks, monkeypatch: pytest.MonkeyPatch, tmp_path: Path, +): + """Terminated -> run_finished{terminated}, then die by SIGTERM (patched).""" + m = serial_main_mocks + _two_devices_interrupted_on_second(m, automate_observation.Terminated()) + events_path = tmp_path / "events.jsonl" + _set_argv(monkeypatch, "--events", str(events_path)) + before = signal.getsignal(signal.SIGTERM) + + with mock.patch("automate_observation._die_by_sigterm") as die: + automate_observation.main() + die.assert_called_once_with(mock.ANY) + assert signal.getsignal(signal.SIGTERM) == before + + events = _read_events(events_path) + finished = {e["device"]: e for e in _of_type(events, "device_finished")} + assert finished["DEV2"]["status"] == "failed" + assert finished["DEV2"]["error"] == {"reason": "terminated", + "message": "Terminated."} + run_finished = events[-1] + assert run_finished["type"] == "run_finished" + assert run_finished["exit_code"] == 143 + assert run_finished["ok"] is False + assert run_finished["error"] == {"reason": "terminated", + "message": "Terminated by SIGTERM."} + assert run_finished["summary"] == { + "DEV1": {"status": "success", "results_dir": "/tmp/out/DEV1"}, + "DEV2": {"status": "failed"}, + } + + +_SIGTERM_DRIVER = textwrap.dedent(""" + import sys, time + from unittest import mock + sys.path.insert(0, {script_dir!r}) + import automate_observation as ao + + def dev(serial): + d = mock.Mock(); d.serial_number = serial; d.unauthorized = False + return d + + waits = iter(["/sdcard/hubble/results", None]) + def wait_for_results(*unused): + result = next(waits) + if result is None: + time.sleep(60) # DEV2 blocks here until the test sends SIGTERM + return result + + patches = dict( + supported_platform=True, verify_hubble=True, adb_installed=True, + is_hubble_installed=False, is_xiaomi_phone=False, install_hubble=True, + clear_logcat=None, launch_hubble=True, extract_selinux_policies=None) + for name, value in patches.items(): + mock.patch.object(ao, name, return_value=value).start() + adb = mock.patch.object(ao, "AdbWrapper").start() + adb.start_server.return_value = True + adb.devices.return_value = [dev("DEV1"), dev("DEV2")] + mock.patch.object(ao, "wait_for_results", + side_effect=wait_for_results).start() + mock.patch.object(ao, "extract_results_and_apks", + return_value="/tmp/out/DEV1").start() + sys.argv = ["automate_observation.py", "-H", "x.apk", + "--events", {events_path!r}] + ao.main() +""") + + +def test_events_sigterm_in_real_process(tmp_path: Path): + """A real SIGTERM mid-run: run_finished is written, exit status is -SIGTERM.""" + events_path = tmp_path / "events.jsonl" + driver = tmp_path / "driver.py" + driver.write_text(_SIGTERM_DRIVER.format(script_dir=SCRIPT_DIR, + events_path=str(events_path))) + proc = subprocess.Popen([sys.executable, str(driver)], + stdout=subprocess.PIPE, stderr=subprocess.PIPE, + text=True) + try: + deadline = time.monotonic() + 30 + while time.monotonic() < deadline: + text = events_path.read_text() if events_path.exists() else "" + if '"step": "wait_for_results", "state": "started", "device": "DEV2"' in text: + break + if proc.poll() is not None: + pytest.fail("driver exited early: " + proc.stderr.read()) + time.sleep(0.05) + else: + pytest.fail("DEV2 never reached wait_for_results") + proc.send_signal(signal.SIGTERM) + _, stderr = proc.communicate(timeout=30) + finally: + if proc.poll() is None: + proc.kill() + + assert proc.returncode == -signal.SIGTERM, stderr + events = _read_events(events_path) + assert _steps(events, "DEV2")[-1] == ("wait_for_results", "failed") + finished = {e["device"]: e for e in _of_type(events, "device_finished")} + assert finished["DEV2"]["error"] == {"reason": "terminated", + "message": "Terminated."} + run_finished = events[-1] + assert run_finished["type"] == "run_finished" + assert run_finished["exit_code"] == 143 + assert run_finished["error"]["reason"] == "terminated" + assert run_finished["summary"] == { + "DEV1": {"status": "success", "results_dir": "/tmp/out/DEV1"}, + "DEV2": {"status": "failed"}, + } + assert "Terminated by SIGTERM." in stderr + + +def test_main_events_unopenable_path_exits_1( + serial_main_mocks, monkeypatch: pytest.MonkeyPatch, tmp_path: Path, +): + bad_path = tmp_path / "no" / "such" / "dir" / "events.jsonl" + _set_argv(monkeypatch, "--events", str(bad_path)) + with pytest.raises(SystemExit) as exc_info: + automate_observation.main() + assert exc_info.value.code == 1 + serial_main_mocks["supported_platform"].assert_not_called() + + +def test_main_without_events_writes_nothing( + serial_main_mocks, monkeypatch: pytest.MonkeyPatch, tmp_path: Path, +): + serial_main_mocks["AdbWrapper"].devices.return_value = [ + _make_mock_device("DEV1")] + monkeypatch.chdir(tmp_path) + _set_argv(monkeypatch) + automate_observation.main() + assert list(tmp_path.iterdir()) == [] + + +# --- --events - (stdout) in a real process ----------------------------------------- + + +def test_events_to_stdout_end_to_end_is_pure_json(tmp_path: Path): + """A real run that stops early: stdout must carry only JSON Lines.""" + not_an_apk = tmp_path / "hubble.txt" + not_an_apk.write_text("") + proc = subprocess.run( + [sys.executable, os.path.join(SCRIPT_DIR, "automate_observation.py"), + "-H", str(not_an_apk), "--events", "-"], + capture_output=True, text=True, timeout=60, check=False) + + assert proc.returncode == 0, proc.stderr + events = [json.loads(line) for line in proc.stdout.splitlines()] + assert [e["type"] for e in events] == [ + "run_started", "step", "step", "run_finished"] + assert _steps(events) == [("verify_hubble", "started"), + ("verify_hubble", "failed")] + assert events[-1]["error"]["reason"] == "invalid_hubble_apk" + # Human-readable logging still goes to stderr. + assert ".apk extension" in proc.stderr + + +def test_open_event_stream_dash_moves_other_stdout_writes_to_stderr(): + """print(), input() prompts and child processes must not pollute stdout.""" + code = textwrap.dedent(""" + import subprocess, sys + sys.path.insert(0, {script_dir!r}) + import automate_observation as ao + emitter = ao.EventEmitter(ao.open_event_stream("-")) + print("noise from print") + subprocess.run([sys.executable, "-c", "print('noise from child')"]) + emitter.emit("hello") + emitter.close() + """).format(script_dir=SCRIPT_DIR) + proc = subprocess.run([sys.executable, "-c", code], capture_output=True, + text=True, timeout=60, check=False) + + assert proc.returncode == 0, proc.stderr + (line,) = proc.stdout.splitlines() + assert json.loads(line)["type"] == "hello" + assert "noise from print" in proc.stderr + assert "noise from child" in proc.stderr + + +if __name__ == "__main__": + sys.exit(pytest.main([__file__]))