1#!/usr/bin/env python3
2"""Run the provider lab on an explicitly selected disposable emulator."""
3import argparse
4import hashlib
5import json
6from pathlib import Path
7import subprocess
8
9
10def main():
11 parser = argparse.ArgumentParser(description=__doc__)
12 parser.add_argument("--adb", type=Path, required=True)
13 parser.add_argument("--serial", required=True)
14 parser.add_argument("--build", type=Path, required=True)
15 parser.add_argument("--output", type=Path, required=True)
16 args = parser.parse_args()
17 output = args.output.resolve()
18 output.mkdir(parents=True, exist_ok=False)
19 steps = []
20
21 def adb(*command, timeout=120):
22 result = subprocess.run([str(args.adb), "-s", args.serial, *command],
23 capture_output=True, text=True, timeout=timeout)
24 steps.append({"command": list(command), "exitCode": result.returncode,
25 "stdout": result.stdout, "stderr": result.stderr})
26 (output / "commands.json").write_text(json.dumps(steps, indent=2) + "\n")
27 result.check_returncode()
28 return result.stdout
29
30 if adb("shell", "getprop", "ro.kernel.qemu").strip() != "1":
31 raise RuntimeError("This lab only runs on a disposable emulator")
32 manifest = json.loads((args.build / "manifest.json").read_text())
33 for name, expected in manifest["apkSha256"].items():
34 if hashlib.sha256((args.build / name).read_bytes()).hexdigest() != expected:
35 raise RuntimeError(f"APK hash changed: {name}")
36 reports = {}
37 packages = ("net.paperclover.snowbound.providerlab", "net.paperclover.snowbound.providerprobe")
38 try:
39 for name in ("provider", "client"):
40 adb("install", "-t", str((args.build / (name + ".apk")).resolve()))
41 for phase in ("probe", "reopen"):
42 adb("shell", "am", "instrument", "-w", "-r", "-e", "phase", phase,
43 "net.paperclover.snowbound.providerprobe/.Probe")
44 report = json.loads(adb("shell", "run-as", packages[1], "cat", "files/" + phase + ".json"))
45 reports[phase] = report
46 (output / (phase + ".json")).write_text(json.dumps(report, indent=2) + "\n")
47 if not report["passed"]:
48 raise RuntimeError(f"{phase} failed: {report.get('failure')}")
49 adb("shell", "am", "force-stop", packages[1])
50 if reports["probe"]["pid"] == reports["reopen"]["pid"]:
51 raise RuntimeError("Second phase did not use a fresh client process")
52 result = {"passed": True, "serial": args.serial, "build": manifest,
53 "reportsSha256": {name + ".json": hashlib.sha256((output / (name + ".json")).read_bytes()).hexdigest()
54 for name in reports},
55 "scope": "Synthetic local DocumentsProvider; no commercial provider, picker UI, lock/CAS, network upload, OneNote, or production storage integration"}
56 (output / "verification.json").write_text(json.dumps(result, indent=2) + "\n")
57 print(json.dumps(result, indent=2))
58 finally:
59 errors = []
60 for package in reversed(packages):
61 try:
62 adb("uninstall", package, timeout=15)
63 except Exception as error:
64 errors.append({"package": package, "operation": "uninstall", "error": repr(error)})
65 retained = []
66 unknown = []
67 for package in packages:
68 try:
69 if adb("shell", "pm", "list", "packages", package, timeout=10).strip():
70 retained.append(package)
71 except Exception as error:
72 unknown.append(package)
73 errors.append({"package": package, "operation": "presence", "error": repr(error)})
74 cleanup = {"packagesAbsent": not (retained or unknown), "retained": retained,
75 "unknown": unknown, "errors": errors}
76 (output / "cleanup.json").write_text(json.dumps(cleanup, indent=2) + "\n")
77 if errors or retained or unknown:
78 raise RuntimeError(f"Lab cleanup failed: {cleanup}")
79
80
81if __name__ == "__main__":
82 main()