mirror of
https://github.com/Droid-Deck/DroidDeck.git
synced 2026-11-12 11:31:08 +00:00
351 lines
16 KiB
Python
351 lines
16 KiB
Python
"""droiddeckctl and droiddeck-scenario against a fake adb: the JSON contract and exit codes."""
|
|
import base64
|
|
import contextlib
|
|
import importlib.machinery
|
|
import importlib.util
|
|
import io
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import stat
|
|
import sys
|
|
import tempfile
|
|
import textwrap
|
|
import unittest
|
|
|
|
TOOLS = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(TOOLS))
|
|
|
|
|
|
def load(name):
|
|
loader = importlib.machinery.SourceFileLoader(name.replace("-", "_"), str(TOOLS / name))
|
|
spec = importlib.util.spec_from_loader(loader.name, loader)
|
|
module = importlib.util.module_from_spec(spec)
|
|
loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
ctl = load("droiddeckctl")
|
|
scenario = load("droiddeck-scenario")
|
|
|
|
FAKE_ADB = textwrap.dedent('''\
|
|
#!/usr/bin/env python3
|
|
import base64, json, os, sys
|
|
config_path = os.environ["FAKE_ADB_CONFIG"]
|
|
config = json.load(open(config_path))
|
|
args = sys.argv[1:]
|
|
if args[:1] == ["-s"]:
|
|
args = args[2:]
|
|
entry = {"args": args}
|
|
if args[:1] == ["get-state"]:
|
|
print("device"); sys.exit(0)
|
|
if args[:3] == ["shell", "content", "call"]:
|
|
method = args[args.index("--method") + 1]
|
|
request = None
|
|
if "--extra" in args:
|
|
request = json.loads(base64.b64decode(args[args.index("--extra") + 1].split(":", 2)[2]))
|
|
entry.update(method=method, request=request)
|
|
answers = config["provider"].get(method, {"ok": False, "error": {"code": "UNKNOWN_COMMAND", "message": method}})
|
|
if isinstance(answers, list):
|
|
count = config.setdefault("counts", {}).get(method, 0)
|
|
config["counts"][method] = count + 1
|
|
answers = answers[min(count, len(answers) - 1)]
|
|
json.dump(config, open(config_path, "w"))
|
|
print("Result: Bundle[{json=" + json.dumps(answers) + "}]")
|
|
elif args[:3] == ["shell", "dumpsys", "display"]:
|
|
print(config.get("dumpsys_display", ""))
|
|
elif args[:3] == ["shell", "dumpsys", "power"]:
|
|
print(" mWakefulness=Awake")
|
|
elif args[:2] == ["exec-out", "screencap"]:
|
|
sys.stdout.buffer.write(b"\\x89PNG fake")
|
|
with open(config["log"], "a") as log:
|
|
log.write(json.dumps(entry) + "\\n")
|
|
''')
|
|
|
|
DUMPSYS_DISPLAY = textwrap.dedent('''\
|
|
Display Devices: size=2
|
|
DisplayDeviceInfo{"Built-in Screen": uniqueId="local:111", 1080 x 1920}
|
|
DisplayDeviceInfo{"Screen-2": uniqueId="local:222", 1080 x 1240}
|
|
Logical Displays: size=2
|
|
Display 0:
|
|
mDisplayId=0
|
|
mPrimaryDisplayDevice=Built-in Screen
|
|
mBaseDisplayInfo=DisplayInfo{"Built-in Screen", displayId 0", real 1080 x 1920, x}
|
|
Display 4:
|
|
mDisplayId=4
|
|
mPrimaryDisplayDevice=Screen-2
|
|
mBaseDisplayInfo=DisplayInfo{"Screen-2", displayId 4", real 1080 x 1240, x}
|
|
''')
|
|
|
|
|
|
def state(phase="READY", focused=None, focusable=()):
|
|
focus = None if focused is None else {"focusedApp": focused, "focusableApps": list(focusable),
|
|
"baselayerAppIds": [], "seq": 1, "t": 0}
|
|
return {"ok": True, "schema": 2, "session": {"id": "s1", "phase": phase, "running": phase == "READY",
|
|
"focus": focus, "failure": None}}
|
|
|
|
|
|
class FakeAdbTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.dir = Path(self.temp.name)
|
|
self.adb = self.dir / "adb"
|
|
self.adb.write_text(FAKE_ADB)
|
|
self.adb.chmod(self.adb.stat().st_mode | stat.S_IEXEC)
|
|
self.config = self.dir / "config.json"
|
|
self.log = self.dir / "log.jsonl"
|
|
self.configure({})
|
|
os.environ["FAKE_ADB_CONFIG"] = str(self.config)
|
|
self.addCleanup(os.environ.pop, "FAKE_ADB_CONFIG", None)
|
|
|
|
def configure(self, provider, **extra):
|
|
self.config.write_text(json.dumps(dict({"provider": provider, "log": str(self.log)}, **extra)))
|
|
|
|
def run_ctl(self, *argv):
|
|
out = io.StringIO()
|
|
with contextlib.redirect_stdout(out), contextlib.redirect_stderr(io.StringIO()):
|
|
code = ctl.main(["--adb", str(self.adb), "--serial", "fake-1", *argv])
|
|
return code, json.loads(out.getvalue())
|
|
|
|
def calls(self, method=None):
|
|
if not self.log.exists():
|
|
return []
|
|
entries = [json.loads(line) for line in self.log.read_text().splitlines()]
|
|
return [e for e in entries if method is None or e.get("method") == method]
|
|
|
|
|
|
class DroiddeckctlTest(FakeAdbTest):
|
|
def test_oversized_requests_are_refused_before_adb(self):
|
|
with self.assertRaises(ctl.ControlError) as caught:
|
|
ctl.provider_call(str(self.adb), "fake-1", "guest", request={"stdin": "x" * ctl.REQUEST_CAP})
|
|
self.assertEqual("INVALID_REQUEST", caught.exception.code)
|
|
self.assertFalse(self.calls())
|
|
|
|
def test_state_passes_through_with_the_serial(self):
|
|
self.configure({"state": state()})
|
|
code, out = self.run_ctl("state")
|
|
self.assertEqual(0, code)
|
|
self.assertEqual("READY", out["session"]["phase"])
|
|
self.assertEqual("fake-1", out["deviceSerial"])
|
|
|
|
def test_launch_sends_the_app_id_and_waits_for_focus(self):
|
|
self.configure({
|
|
"launch": {"ok": True, "command": "launch", "appId": "489830"},
|
|
"state": [state(focused=769, focusable=[769]), state(focused=489830, focusable=[769])],
|
|
})
|
|
code, out = self.run_ctl("launch", "489830", "--wait", "--timeout", "1")
|
|
self.assertEqual(5, code)
|
|
self.assertEqual("TIMEOUT", out["error"]["code"])
|
|
self.assertIn("focused app 489830", out["error"]["message"])
|
|
self.assertEqual({"appId": "489830"}, self.calls("launch")[0]["request"])
|
|
|
|
self.configure({
|
|
"launch": {"ok": True, "command": "launch", "appId": "489830"},
|
|
"state": [state(focused=769, focusable=[769]), state(focused=489830, focusable=[489830, 769])],
|
|
})
|
|
code, out = self.run_ctl("launch", "489830", "--wait", "--timeout", "5")
|
|
self.assertEqual(0, code)
|
|
self.assertEqual(489830, out["focus"]["focusedApp"])
|
|
|
|
def test_start_reuse_keeps_a_running_session(self):
|
|
running = state(focused=769)
|
|
running["session"]["mode"] = "steam"
|
|
self.configure({"state": running})
|
|
code, out = self.run_ctl("start", "steam", "--reuse", "--wait")
|
|
self.assertEqual(0, code)
|
|
self.assertTrue(out["reused"])
|
|
self.assertFalse([c for c in self.calls() if c["args"][:2] == ["shell", "am"]])
|
|
|
|
def test_start_without_reuse_refuses_a_running_session(self):
|
|
running = state()
|
|
running["session"]["mode"] = "steam"
|
|
self.configure({"state": running})
|
|
code, out = self.run_ctl("start", "steam")
|
|
self.assertEqual(4, code)
|
|
self.assertEqual("SESSION_ACTIVE", out["error"]["code"])
|
|
|
|
def test_wait_game_ignores_the_client(self):
|
|
self.configure({"state": [state(focused=769), state(focused=752590, focusable=[752590])]})
|
|
code, out = self.run_ctl("wait", "game", "--timeout", "5")
|
|
self.assertEqual(0, code)
|
|
self.assertEqual(752590, out["session"]["focus"]["focusedApp"])
|
|
|
|
def test_a_failed_session_ends_a_wait(self):
|
|
failed = state(phase="FAILED")
|
|
failed["session"]["failure"] = {"code": "GUEST_EXIT", "message": "boom"}
|
|
self.configure({"state": failed})
|
|
code, out = self.run_ctl("wait", "focused", "1", "--timeout", "5")
|
|
self.assertEqual(4, code)
|
|
self.assertEqual("GUEST_EXIT", out["error"]["code"])
|
|
|
|
def test_disabled_agent_commands_exit_7(self):
|
|
self.configure({"guest": {"ok": False, "error": {"code": "AGENT_COMMANDS_DISABLED", "message": "off"}}})
|
|
code, out = self.run_ctl("guest", "--", "true")
|
|
self.assertEqual(7, code)
|
|
self.assertEqual("AGENT_COMMANDS_DISABLED", out["error"]["code"])
|
|
|
|
def test_run_checks_access_before_launching_an_activity(self):
|
|
self.configure({"access": {"ok": True, "commands": False}})
|
|
code, out = self.run_ctl("run", "/usr/bin/sh", "--", "-c", "true")
|
|
self.assertEqual(7, code)
|
|
self.assertEqual("AGENT_COMMANDS_DISABLED", out["error"]["code"])
|
|
self.assertIn("Setup > Diagnostics > Debugging tools", out["error"]["message"])
|
|
self.assertEqual(["access"], [c.get("method") for c in self.calls()])
|
|
|
|
def test_override_checks_names_before_pushing_any_file(self):
|
|
self.configure({"access": {"ok": True, "commands": True, "inbox": "/tmp/inbox"}})
|
|
source = self.dir / "binary"
|
|
source.write_bytes(b"test")
|
|
code, out = self.run_ctl("override", "../outside", str(source))
|
|
self.assertEqual(2, code)
|
|
self.assertEqual("INVALID_NAME", out["error"]["code"])
|
|
self.assertEqual(["access"], [c.get("method") for c in self.calls()])
|
|
|
|
def test_guest_request_and_check(self):
|
|
self.configure({"guest": {"ok": True, "exitCode": 3, "stdout": {"text": ""}, "stderr": {"text": "nope"}}})
|
|
code, out = self.run_ctl("guest", "--timeout", "7", "--env", "A=1", "--check", "--", "xprop", "-root")
|
|
self.assertEqual(4, code)
|
|
self.assertEqual("GUEST_EXIT", out["error"]["code"])
|
|
self.assertEqual({"argv": ["xprop", "-root"], "timeout": 7.0, "env": {"A": "1"}},
|
|
self.calls("guest")[0]["request"])
|
|
|
|
def test_screenshot_resolves_the_logical_display(self):
|
|
self.configure({}, dumpsys_display=DUMPSYS_DISPLAY)
|
|
target = self.dir / "shot.png"
|
|
code, out = self.run_ctl("screenshot", str(target), "--display", "4")
|
|
self.assertEqual(0, code)
|
|
self.assertTrue(target.read_bytes().startswith(b"\x89PNG"))
|
|
shot = [c for c in self.calls() if c["args"][:2] == ["exec-out", "screencap"]][0]
|
|
self.assertEqual(["exec-out", "screencap", "-p", "-d", "222"], shot["args"])
|
|
|
|
def test_displays(self):
|
|
self.configure({}, dumpsys_display=DUMPSYS_DISPLAY)
|
|
code, out = self.run_ctl("displays")
|
|
self.assertEqual(0, code)
|
|
self.assertEqual([0, 4], [d["display"] for d in out["displays"]])
|
|
self.assertEqual("222", out["displays"][1]["physicalId"])
|
|
self.assertEqual([1080, 1240], out["displays"][1]["size"])
|
|
|
|
def test_unknown_display_is_an_error(self):
|
|
self.configure({}, dumpsys_display=DUMPSYS_DISPLAY)
|
|
code, out = self.run_ctl("screenshot", str(self.dir / "x.png"), "--display", "9")
|
|
self.assertEqual(2, code)
|
|
self.assertEqual("NO_SUCH_DISPLAY", out["error"]["code"])
|
|
|
|
def test_input_requests(self):
|
|
self.configure({"input": {"ok": True, "command": "input"}})
|
|
self.run_ctl("input", "tap", "0.5", "0.25", "--space", "fraction")
|
|
self.run_ctl("input", "key", "enter", "--mod", "CTRL")
|
|
self.run_ctl("input", "pad", "a", "--hold", "50")
|
|
requests = [c["request"] for c in self.calls("input")]
|
|
self.assertEqual({"type": "tap", "x": 0.5, "y": 0.25, "space": "fraction", "holdMs": 60}, requests[0])
|
|
self.assertEqual({"type": "key", "key": "enter", "action": "tap", "modifiers": ["CTRL"], "holdMs": 40}, requests[1])
|
|
self.assertEqual({"type": "pad", "button": "a", "action": "tap", "holdMs": 50}, requests[2])
|
|
|
|
def test_ui_click_by_tag_or_match(self):
|
|
self.configure({"ui": {"ok": True, "command": "ui"}})
|
|
self.run_ctl("ui", "click", "rail-steam")
|
|
self.run_ctl("ui", "click", "--match", "Play", "--index", "1")
|
|
requests = [c["request"] for c in self.calls("ui")]
|
|
self.assertEqual({"op": "click", "index": 0, "tag": "rail-steam"}, requests[0])
|
|
self.assertEqual({"op": "click", "index": 1, "match": "Play"}, requests[1])
|
|
|
|
def test_cdp_targets_and_eval_requests(self):
|
|
self.configure({"cdp": {"ok": True, "targets": []}})
|
|
self.assertEqual(0, self.run_ctl("cdp", "targets")[0])
|
|
self.assertEqual(0, self.run_ctl("cdp", "eval", "1+1", "--target", "Big Picture")[0])
|
|
requests = [c["request"] for c in self.calls("cdp")]
|
|
self.assertEqual({"op": "targets"}, requests[0])
|
|
self.assertEqual({"op": "eval", "expression": "1+1", "awaitPromise": True, "timeout": 15.0,
|
|
"target": "Big Picture"}, requests[1])
|
|
|
|
def test_a_bug_still_answers_json(self):
|
|
self.configure({"state": {"ok": True, "session": "not an object"}})
|
|
code, out = self.run_ctl("wait", "game", "--timeout", "1")
|
|
self.assertEqual(2, code)
|
|
self.assertEqual("INTERNAL_ERROR", out["error"]["code"])
|
|
|
|
def test_wake(self):
|
|
code, out = self.run_ctl("wake")
|
|
self.assertEqual(0, code)
|
|
self.assertEqual("Awake", out["wakefulness"])
|
|
self.assertIn(["shell", "input", "keyevent", "KEYCODE_WAKEUP"], [c["args"] for c in self.calls()])
|
|
|
|
def test_override_pushes_to_the_inbox_first(self):
|
|
self.configure({
|
|
"access": {"ok": True, "commands": True, "inbox": "/sdcard/Android/data/x/files/agent-inbox"},
|
|
"override": {"ok": True, "installed": "/opt/droiddeck-agent/bin/gamescope"},
|
|
})
|
|
binary = self.dir / "gamescope"
|
|
binary.write_bytes(b"\x7fELF")
|
|
code, out = self.run_ctl("override", "gamescope", str(binary))
|
|
self.assertEqual(0, code)
|
|
push = [c for c in self.calls() if c["args"][:1] == ["push"]][0]
|
|
self.assertEqual("/sdcard/Android/data/x/files/agent-inbox/gamescope", push["args"][2])
|
|
self.assertEqual({"name": "gamescope"}, self.calls("override")[0]["request"])
|
|
|
|
def test_override_refused_before_pushing_when_commands_are_off(self):
|
|
self.configure({"access": {"ok": True, "commands": False, "inbox": "/x"}})
|
|
binary = self.dir / "gamescope"
|
|
binary.write_bytes(b"x")
|
|
code, out = self.run_ctl("override", "gamescope", str(binary))
|
|
self.assertEqual(7, code)
|
|
self.assertFalse([c for c in self.calls() if c["args"][:1] == ["push"]])
|
|
|
|
|
|
class ScenarioTest(FakeAdbTest):
|
|
def write(self, body):
|
|
path = self.dir / "scenario.json"
|
|
path.write_text(json.dumps(body))
|
|
return path
|
|
|
|
def run_scenario(self, path, *extra):
|
|
out = io.StringIO()
|
|
with contextlib.redirect_stdout(out), contextlib.redirect_stderr(io.StringIO()):
|
|
code = scenario.main([str(path), "--adb", str(self.adb), "--serial", "fake-1",
|
|
"--out", str(self.dir / "runs"), *extra])
|
|
return code, json.loads(out.getvalue())
|
|
|
|
def test_variables_asserts_and_repeats(self):
|
|
self.configure({"state": state(focused=489830, focusable=[489830])})
|
|
path = self.write({
|
|
"name": "t", "vars": {"appId": 1}, "repeat": 2,
|
|
"steps": [["state"], {"assert": "session.focus.focusedApp", "equals": "${appId}"}],
|
|
})
|
|
code, report = self.run_scenario(path, "--var", "appId=489830")
|
|
self.assertEqual(0, code)
|
|
self.assertEqual(2, report["passed"])
|
|
output = Path(report["out"])
|
|
self.assertEqual(0o700, output.stat().st_mode & 0o777)
|
|
for artifact in output.rglob("*.json"):
|
|
self.assertEqual(0o600, artifact.stat().st_mode & 0o777, str(artifact))
|
|
for run in report["runs"]:
|
|
self.assertEqual(0o700, Path(run["folder"]).stat().st_mode & 0o777)
|
|
|
|
def test_a_failed_assert_fails_the_run_and_collects_evidence(self):
|
|
self.configure({"state": state(focused=769, focusable=[769]), "focus": {"ok": True, "focusedApp": 769}},
|
|
dumpsys_display=DUMPSYS_DISPLAY)
|
|
path = self.write({"name": "t", "repeat": 3,
|
|
"steps": [["state"], {"assert": "session.focus.focusedApp", "equals": 489830}]})
|
|
code, report = self.run_scenario(path)
|
|
self.assertEqual(1, code)
|
|
self.assertEqual(1, len(report["runs"]))
|
|
self.assertIn("is 769", report["runs"][0]["error"])
|
|
evidence = json.loads((Path(report["runs"][0]["folder"]) / "evidence.json").read_text())
|
|
self.assertEqual(769, evidence["focus"]["focusedApp"])
|
|
|
|
def test_after_steps_run_after_a_failure(self):
|
|
self.configure({"state": state(focused=769), "quit": {"ok": True, "remaining": []}})
|
|
path = self.write({"name": "t", "repeat": 2, "vars": {"appId": 7},
|
|
"steps": [["state"], {"assert": "session.focus.focusedApp", "equals": 7}],
|
|
"after": [["quit", "${appId}"]]})
|
|
code, report = self.run_scenario(path, "--keep-going")
|
|
self.assertEqual(1, code)
|
|
self.assertEqual(2, len(report["runs"]))
|
|
self.assertEqual([{"appId": 7, "timeout": 15.0}] * 2, [c["request"] for c in self.calls("quit")])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|