generated from exteraGramPlugins/pluginBase
Initial commit
This commit is contained in:
exteraGramPlugins committed
commit
a3b0b264c1
27 files changed
+1699
No files matched your search
Executable
+345
@@ -0,0 +1,345 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Bundle a multi-file exteraGram plugin (plugin.json + src/) into a single .plugin file.
|
||||
|
||||
Usage: python3 tools/bundle.py [--dev|--release] [--out DIR] [--check]
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import ast
|
||||
import base64
|
||||
import json
|
||||
import re
|
||||
import sys
|
||||
import zlib
|
||||
from pathlib import Path
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parent.parent
|
||||
SRC_DIR = REPO_ROOT / "src"
|
||||
PLUGIN_JSON = REPO_ROOT / "plugin.json"
|
||||
|
||||
RECOGNIZED_KEYS = {
|
||||
"id",
|
||||
"name",
|
||||
"description",
|
||||
"author",
|
||||
"version",
|
||||
"icon",
|
||||
"app_version",
|
||||
"sdk_version",
|
||||
"requirements",
|
||||
}
|
||||
REQUIRED_KEYS = {"id", "name"}
|
||||
ID_RE = re.compile(r"^[A-Za-z][A-Za-z0-9_-]{1,31}$")
|
||||
|
||||
KEY_TO_DUNDER = {
|
||||
"id": "__id__",
|
||||
"name": "__name__",
|
||||
"description": "__description__",
|
||||
"author": "__author__",
|
||||
"version": "__version__",
|
||||
"icon": "__icon__",
|
||||
"app_version": "__app_version__",
|
||||
"sdk_version": "__sdk_version__",
|
||||
"requirements": "__requirements__",
|
||||
}
|
||||
METADATA_DUNDERS = set(KEY_TO_DUNDER.values()) | {"__min_version__"}
|
||||
|
||||
RUNTIME_PREAMBLE = '''\
|
||||
import base64 as _b64, builtins as _bi, json as _j, linecache as _lc, types as _t, zlib as _z
|
||||
|
||||
_BUNDLE = _j.loads(_z.decompress(_b64.b64decode(_BUNDLE_BLOB)).decode("utf-8"))
|
||||
_MODS = _BUNDLE["modules"]
|
||||
_LOADED = {}
|
||||
_ROOTS = {n.partition(".")[0] for n in _MODS}
|
||||
_REAL_IMPORT = _bi.__import__
|
||||
|
||||
|
||||
def _bundle_register_source(info):
|
||||
lines = info["src"].splitlines(True)
|
||||
_lc.cache[info["file"]] = (len(info["src"]), None, lines, info["file"])
|
||||
|
||||
|
||||
def _bundle_load(name):
|
||||
mod = _LOADED.get(name)
|
||||
if mod is not None:
|
||||
return mod
|
||||
info = _MODS.get(name)
|
||||
if info is None:
|
||||
raise ImportError("no bundled module named %r" % name)
|
||||
parent = name.rpartition(".")[0]
|
||||
if parent:
|
||||
_bundle_load(parent)
|
||||
mod = _t.ModuleType(name)
|
||||
mod.__file__ = info["file"]
|
||||
mod.__package__ = name if info["pkg"] else parent
|
||||
if info["pkg"]:
|
||||
mod.__path__ = []
|
||||
builtins_ns = dict(vars(_bi))
|
||||
builtins_ns["__import__"] = _bundle_import
|
||||
mod.__dict__["__builtins__"] = builtins_ns
|
||||
_LOADED[name] = mod
|
||||
_bundle_register_source(info)
|
||||
try:
|
||||
exec(compile(info["src"], info["file"], "exec"), mod.__dict__)
|
||||
except BaseException:
|
||||
_LOADED.pop(name, None)
|
||||
raise
|
||||
if parent:
|
||||
setattr(_LOADED[parent], name.rpartition(".")[2], mod)
|
||||
return mod
|
||||
|
||||
|
||||
def _bundle_import(name, globals=None, locals=None, fromlist=(), level=0):
|
||||
if level:
|
||||
pkg = (globals or {}).get("__package__") or ""
|
||||
parts = pkg.split(".") if pkg else []
|
||||
if level > 1:
|
||||
parts = parts[: 1 - level]
|
||||
if not parts:
|
||||
return _REAL_IMPORT(name, globals, locals, fromlist, level)
|
||||
full = ".".join(parts) + ("." + name if name else "")
|
||||
else:
|
||||
full = name
|
||||
if full.partition(".")[0] not in _ROOTS or full not in _MODS:
|
||||
return _REAL_IMPORT(name, globals, locals, fromlist, level)
|
||||
mod = _bundle_load(full)
|
||||
if not fromlist:
|
||||
return mod if level else _bundle_load(full.partition(".")[0])
|
||||
for item in fromlist:
|
||||
if item != "*" and not hasattr(mod, item) and full + "." + item in _MODS:
|
||||
_bundle_load(full + "." + item)
|
||||
return mod
|
||||
'''
|
||||
|
||||
ENTRY_EXECUTION = '''\
|
||||
_ENTRY = _MODS[_BUNDLE["entry"]]
|
||||
_bundle_register_source(_ENTRY)
|
||||
_bundle_builtins = dict(vars(_bi))
|
||||
_bundle_builtins["__import__"] = _bundle_import
|
||||
globals()["__builtins__"] = _bundle_builtins
|
||||
exec(compile(_ENTRY["src"], _ENTRY["file"], "exec"), globals())
|
||||
'''
|
||||
|
||||
|
||||
class ValidationError(Exception):
|
||||
pass
|
||||
|
||||
|
||||
def load_metadata() -> dict:
|
||||
if not PLUGIN_JSON.is_file():
|
||||
raise ValidationError(f"plugin.json not found at {PLUGIN_JSON}")
|
||||
try:
|
||||
text = PLUGIN_JSON.read_text(encoding="utf-8")
|
||||
except OSError as exc:
|
||||
raise ValidationError(f"cannot read plugin.json: {exc}")
|
||||
try:
|
||||
data = json.loads(text)
|
||||
except json.JSONDecodeError as exc:
|
||||
raise ValidationError(f"plugin.json is not valid JSON: {exc}")
|
||||
if not isinstance(data, dict):
|
||||
raise ValidationError("plugin.json must be a JSON object")
|
||||
unknown = set(data) - RECOGNIZED_KEYS
|
||||
if unknown:
|
||||
raise ValidationError(
|
||||
"plugin.json has unrecognized keys: " + ", ".join(sorted(unknown))
|
||||
)
|
||||
missing = REQUIRED_KEYS - set(data)
|
||||
if missing:
|
||||
raise ValidationError(
|
||||
"plugin.json is missing required keys: " + ", ".join(sorted(missing))
|
||||
)
|
||||
plugin_id = data["id"]
|
||||
if not isinstance(plugin_id, str) or not ID_RE.match(plugin_id):
|
||||
raise ValidationError(
|
||||
f"plugin.json id {plugin_id!r} is invalid: must be 2-32 chars, start with a "
|
||||
"letter, and contain only A-Za-z0-9_-"
|
||||
)
|
||||
if not isinstance(data["name"], str) or not data["name"]:
|
||||
raise ValidationError("plugin.json name must be a non-empty string")
|
||||
return data
|
||||
|
||||
|
||||
def collect_modules() -> dict:
|
||||
if not SRC_DIR.is_dir():
|
||||
raise ValidationError(f"src/ directory not found at {SRC_DIR}")
|
||||
modules: dict = {}
|
||||
for path in sorted(SRC_DIR.rglob("*")):
|
||||
if path.is_dir():
|
||||
continue
|
||||
rel = path.relative_to(SRC_DIR)
|
||||
if any(part.startswith(".") for part in rel.parts):
|
||||
continue
|
||||
if "__pycache__" in rel.parts:
|
||||
continue
|
||||
if path.suffix == ".pyc":
|
||||
continue
|
||||
if path.suffix != ".py":
|
||||
print(
|
||||
f"warning: skipping non-Python file {path.relative_to(REPO_ROOT)} "
|
||||
"(single-file plugins cannot carry assets)",
|
||||
file=sys.stderr,
|
||||
)
|
||||
continue
|
||||
is_pkg = rel.name == "__init__.py"
|
||||
dotted_parts = rel.parts[:-1] if is_pkg else rel.parts[:-1] + (rel.stem,)
|
||||
name = ".".join(dotted_parts)
|
||||
if not name:
|
||||
raise ValidationError(f"cannot derive a module name for {path}")
|
||||
try:
|
||||
src = path.read_text(encoding="utf-8")
|
||||
except OSError as exc:
|
||||
raise ValidationError(f"cannot read {path}: {exc}")
|
||||
modules[name] = {"src": src, "path": path, "pkg": is_pkg}
|
||||
return modules
|
||||
|
||||
|
||||
def validate_entry(modules: dict) -> None:
|
||||
main_path = SRC_DIR / "main.py"
|
||||
if not main_path.is_file():
|
||||
raise ValidationError("src/main.py does not exist")
|
||||
entry = modules.get("main")
|
||||
if entry is None:
|
||||
raise ValidationError("src/main.py could not be collected as the 'main' module")
|
||||
try:
|
||||
tree = ast.parse(entry["src"], filename=str(main_path))
|
||||
except SyntaxError as exc:
|
||||
raise ValidationError(f"src/main.py has a syntax error: {exc}")
|
||||
|
||||
for node in ast.walk(tree):
|
||||
if isinstance(node, ast.ImportFrom) and node.level > 0:
|
||||
raise ValidationError(
|
||||
"src/main.py must use absolute imports only (found a relative import); "
|
||||
"it is executed at plugin-module top level"
|
||||
)
|
||||
if isinstance(node, (ast.Assign, ast.AnnAssign)):
|
||||
targets = node.targets if isinstance(node, ast.Assign) else [node.target]
|
||||
for target in targets:
|
||||
if isinstance(target, ast.Name) and target.id in METADATA_DUNDERS:
|
||||
raise ValidationError(
|
||||
f"src/main.py must not assign metadata dunder {target.id!r} "
|
||||
"(metadata belongs to plugin.json only)"
|
||||
)
|
||||
|
||||
def _base_names(class_node: ast.ClassDef) -> set:
|
||||
names = set()
|
||||
for base in class_node.bases:
|
||||
if isinstance(base, ast.Name):
|
||||
names.add(base.id)
|
||||
elif isinstance(base, ast.Attribute):
|
||||
names.add(base.attr)
|
||||
return names
|
||||
|
||||
plugin_classes = [
|
||||
node.name
|
||||
for node in tree.body
|
||||
if isinstance(node, ast.ClassDef) and "BasePlugin" in _base_names(node)
|
||||
]
|
||||
if len(plugin_classes) != 1:
|
||||
raise ValidationError(
|
||||
"src/main.py must contain exactly one class inheriting BasePlugin, found "
|
||||
f"{len(plugin_classes)}"
|
||||
)
|
||||
|
||||
|
||||
def validate_modules_syntax(modules: dict) -> None:
|
||||
for name, info in modules.items():
|
||||
try:
|
||||
ast.parse(info["src"], filename=str(info["path"]))
|
||||
except SyntaxError as exc:
|
||||
raise ValidationError(f"{info['path']} has a syntax error: {exc}")
|
||||
|
||||
|
||||
def format_metadata_constants(data: dict) -> str:
|
||||
lines = []
|
||||
for key, dunder in KEY_TO_DUNDER.items():
|
||||
if key not in data:
|
||||
continue
|
||||
value = data[key]
|
||||
if key in ("description", "icon") and value == "":
|
||||
continue
|
||||
if key == "requirements" and value == []:
|
||||
continue
|
||||
lines.append(f"{dunder} = {value!r}")
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def build_payload(modules: dict, release: bool) -> dict:
|
||||
mod_dict = {}
|
||||
for name, info in modules.items():
|
||||
if release:
|
||||
file_value = str(info["path"].relative_to(REPO_ROOT))
|
||||
else:
|
||||
file_value = str(info["path"].resolve())
|
||||
mod_dict[name] = {"src": info["src"], "file": file_value, "pkg": info["pkg"]}
|
||||
return {"modules": mod_dict, "entry": "main"}
|
||||
|
||||
|
||||
def encode_payload(payload: dict) -> str:
|
||||
data = json.dumps(payload, sort_keys=True).encode("utf-8")
|
||||
compressed = zlib.compress(data, 9)
|
||||
return base64.b64encode(compressed).decode("ascii")
|
||||
|
||||
|
||||
def format_blob_literal(blob: str, width: int = 96) -> str:
|
||||
if not blob:
|
||||
return '_BUNDLE_BLOB = ""'
|
||||
lines = [blob[i : i + width] for i in range(0, len(blob), width)]
|
||||
body = "\n".join(f' "{line}"' for line in lines)
|
||||
return "_BUNDLE_BLOB = (\n" + body + "\n)"
|
||||
|
||||
|
||||
def assemble_bundle(data: dict, modules: dict, release: bool) -> str:
|
||||
header = "# Generated by tools/bundle.py from plugin.json + src/ — do not edit; edit the sources.\n"
|
||||
metadata = format_metadata_constants(data)
|
||||
payload = build_payload(modules, release)
|
||||
blob_literal = format_blob_literal(encode_payload(payload))
|
||||
text = "\n".join([header, metadata, "", blob_literal, "", RUNTIME_PREAMBLE, ENTRY_EXECUTION])
|
||||
if not text.endswith("\n"):
|
||||
text += "\n"
|
||||
return text
|
||||
|
||||
|
||||
def parse_args(argv):
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Bundle a multi-file exteraGram plugin into a single .plugin file."
|
||||
)
|
||||
mode = parser.add_mutually_exclusive_group()
|
||||
mode.add_argument(
|
||||
"--dev", action="store_true", help="embed absolute local paths for debugging (default)"
|
||||
)
|
||||
mode.add_argument(
|
||||
"--release", action="store_true", help="embed repo-relative paths for distribution"
|
||||
)
|
||||
parser.add_argument("--out", default="dist", help="output directory (default: dist)")
|
||||
parser.add_argument("--check", action="store_true", help="validate only; do not write output")
|
||||
return parser.parse_args(argv)
|
||||
|
||||
|
||||
def main(argv=None) -> int:
|
||||
args = parse_args(sys.argv[1:] if argv is None else argv)
|
||||
release = bool(args.release)
|
||||
try:
|
||||
data = load_metadata()
|
||||
modules = collect_modules()
|
||||
validate_entry(modules)
|
||||
validate_modules_syntax(modules)
|
||||
except ValidationError as exc:
|
||||
print(f"error: {exc}", file=sys.stderr)
|
||||
return 2
|
||||
|
||||
if args.check:
|
||||
print("ok")
|
||||
return 0
|
||||
|
||||
text = assemble_bundle(data, modules, release)
|
||||
out_dir = Path(args.out)
|
||||
out_dir.mkdir(parents=True, exist_ok=True)
|
||||
out_path = out_dir / f"{data['id']}.plugin"
|
||||
out_path.write_text(text, encoding="utf-8")
|
||||
print(f"wrote {out_path} ({out_path.stat().st_size} bytes)")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Executable
+219
@@ -0,0 +1,219 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Client for the exteraGram on-device plugin dev server.
|
||||
|
||||
Speaks the wire protocol at 127.0.0.1:42690 (after `adb forward tcp:42690 tcp:42690`):
|
||||
consecutive UTF-8 JSON objects, no length framing. Requests are
|
||||
{"@": <command>, "#": <id>, ...}; responses echo "#".
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import socket
|
||||
import sys
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parent.parent
|
||||
PLUGIN_JSON = REPO_ROOT / "plugin.json"
|
||||
|
||||
CONNECT_ATTEMPTS = 5
|
||||
CONNECT_RETRY_DELAY = 1.0
|
||||
SOCKET_TIMEOUT = 1.0
|
||||
RESPONSE_TIMEOUT = 30.0
|
||||
|
||||
UNREACHABLE_MESSAGE = (
|
||||
"dev server unreachable — run scripts/adb-connect.sh, and enable exteraGram "
|
||||
"Preferences > Plugins > plugin engine + developer mode"
|
||||
)
|
||||
|
||||
|
||||
def default_plugin_id() -> str | None:
|
||||
try:
|
||||
data = json.loads(PLUGIN_JSON.read_text(encoding="utf-8"))
|
||||
except (OSError, json.JSONDecodeError):
|
||||
return None
|
||||
return data.get("id")
|
||||
|
||||
|
||||
class DevClient:
|
||||
def __init__(self, host: str, port: int):
|
||||
self.host = host
|
||||
self.port = port
|
||||
self._sock: socket.socket | None = None
|
||||
self._buf = b""
|
||||
self._decoder = json.JSONDecoder()
|
||||
self._next_id = 1
|
||||
|
||||
def connect(self) -> None:
|
||||
last_exc = None
|
||||
for attempt in range(CONNECT_ATTEMPTS):
|
||||
try:
|
||||
sock = socket.create_connection((self.host, self.port), timeout=SOCKET_TIMEOUT)
|
||||
sock.settimeout(SOCKET_TIMEOUT)
|
||||
self._sock = sock
|
||||
return
|
||||
except OSError as exc:
|
||||
last_exc = exc
|
||||
if attempt < CONNECT_ATTEMPTS - 1:
|
||||
time.sleep(CONNECT_RETRY_DELAY)
|
||||
print(UNREACHABLE_MESSAGE, file=sys.stderr)
|
||||
raise SystemExit(1) from last_exc
|
||||
|
||||
def close(self) -> None:
|
||||
if self._sock is not None:
|
||||
self._sock.close()
|
||||
self._sock = None
|
||||
|
||||
def _read_one(self, deadline: float) -> dict:
|
||||
while True:
|
||||
self._buf = self._buf.lstrip()
|
||||
if self._buf:
|
||||
try:
|
||||
text = self._buf.decode("utf-8")
|
||||
except UnicodeDecodeError:
|
||||
text = None
|
||||
if text:
|
||||
try:
|
||||
obj, idx = self._decoder.raw_decode(text)
|
||||
consumed = len(text[:idx].encode("utf-8"))
|
||||
self._buf = self._buf[consumed:]
|
||||
return obj
|
||||
except json.JSONDecodeError:
|
||||
pass
|
||||
if time.monotonic() >= deadline:
|
||||
print("error: timed out waiting for dev server response", file=sys.stderr)
|
||||
raise SystemExit(1)
|
||||
try:
|
||||
chunk = self._sock.recv(4096)
|
||||
except socket.timeout:
|
||||
continue
|
||||
except OSError as exc:
|
||||
print(f"error: connection lost: {exc}", file=sys.stderr)
|
||||
raise SystemExit(1)
|
||||
if not chunk:
|
||||
print("error: dev server closed the connection", file=sys.stderr)
|
||||
raise SystemExit(1)
|
||||
self._buf += chunk
|
||||
|
||||
def request(self, command: str, **fields) -> dict:
|
||||
req_id = self._next_id
|
||||
self._next_id += 1
|
||||
payload = {"@": command, "#": req_id, **fields}
|
||||
self._sock.sendall(json.dumps(payload).encode("utf-8"))
|
||||
deadline = time.monotonic() + RESPONSE_TIMEOUT
|
||||
while True:
|
||||
resp = self._read_one(deadline)
|
||||
if resp.get("#") == req_id:
|
||||
if resp.get("error") or resp.get("errors"):
|
||||
print(json.dumps(resp, indent=2), file=sys.stderr)
|
||||
raise SystemExit(1)
|
||||
return resp
|
||||
# response to a different (stale) request id: keep draining
|
||||
|
||||
|
||||
def cmd_ping(client: DevClient, args) -> int:
|
||||
resp = client.request("ping")
|
||||
print(json.dumps(resp, indent=2))
|
||||
return 0
|
||||
|
||||
|
||||
def cmd_plugins(client: DevClient, args) -> int:
|
||||
resp = client.request("get_plugins")
|
||||
print(json.dumps(resp, indent=2))
|
||||
return 0
|
||||
|
||||
|
||||
def cmd_deploy(client: DevClient, args) -> int:
|
||||
plugin_id = args.id or default_plugin_id()
|
||||
if not plugin_id:
|
||||
print("error: could not determine plugin id (pass --id or fix plugin.json)", file=sys.stderr)
|
||||
return 1
|
||||
try:
|
||||
content = Path(args.file).read_text(encoding="utf-8")
|
||||
except OSError as exc:
|
||||
print(f"error: cannot read {args.file}: {exc}", file=sys.stderr)
|
||||
return 1
|
||||
client.request("write_plugin", plugin_id=plugin_id, content=content)
|
||||
time.sleep(0.3)
|
||||
if args.enable:
|
||||
client.request("enable_plugin", plugin_id=plugin_id)
|
||||
if not args.no_reload:
|
||||
client.request("reload_plugin", plugin_id=plugin_id)
|
||||
print(f"deployed {args.file} as {plugin_id}")
|
||||
return 0
|
||||
|
||||
|
||||
def cmd_simple_id(command: str):
|
||||
def handler(client: DevClient, args) -> int:
|
||||
client.request(command, plugin_id=args.id)
|
||||
print(f"{command} {args.id}: ok")
|
||||
return 0
|
||||
|
||||
return handler
|
||||
|
||||
|
||||
def cmd_debugger(client: DevClient, args) -> int:
|
||||
if args.action == "start":
|
||||
client.request(
|
||||
"start_debugger",
|
||||
host=args.debug_host,
|
||||
port=args.debug_port,
|
||||
platform=args.platform,
|
||||
)
|
||||
print(f"debugger start requested ({args.platform} -> {args.debug_host}:{args.debug_port})")
|
||||
else:
|
||||
client.request("stop_debugger", platform=args.platform)
|
||||
print(f"debugger stop requested ({args.platform})")
|
||||
return 0
|
||||
|
||||
|
||||
def build_parser() -> argparse.ArgumentParser:
|
||||
parser = argparse.ArgumentParser(description="exteraGram plugin dev server client")
|
||||
parser.add_argument("--host", default="127.0.0.1")
|
||||
parser.add_argument("--port", type=int, default=42690)
|
||||
sub = parser.add_subparsers(dest="command", required=True)
|
||||
|
||||
sub.add_parser("ping").set_defaults(func=cmd_ping)
|
||||
sub.add_parser("plugins").set_defaults(func=cmd_plugins)
|
||||
|
||||
p_deploy = sub.add_parser("deploy")
|
||||
p_deploy.add_argument("file")
|
||||
p_deploy.add_argument("--id")
|
||||
p_deploy.add_argument("--enable", action="store_true")
|
||||
p_deploy.add_argument("--no-reload", action="store_true")
|
||||
p_deploy.set_defaults(func=cmd_deploy)
|
||||
|
||||
for name, command in (
|
||||
("enable", "enable_plugin"),
|
||||
("disable", "disable_plugin"),
|
||||
("reload", "reload_plugin"),
|
||||
("remove", "remove_plugin"),
|
||||
):
|
||||
p = sub.add_parser(name)
|
||||
p.add_argument("id")
|
||||
p.set_defaults(func=cmd_simple_id(command))
|
||||
|
||||
p_debugger = sub.add_parser("debugger")
|
||||
p_debugger.add_argument("action", choices=["start", "stop"])
|
||||
p_debugger.add_argument("--platform", choices=["vscode", "pycharm"], default="vscode")
|
||||
p_debugger.add_argument("--debug-host", default="127.0.0.1")
|
||||
p_debugger.add_argument("--debug-port", type=int, default=5678)
|
||||
p_debugger.set_defaults(func=cmd_debugger)
|
||||
|
||||
return parser
|
||||
|
||||
|
||||
def main(argv=None) -> int:
|
||||
parser = build_parser()
|
||||
args = parser.parse_args(sys.argv[1:] if argv is None else argv)
|
||||
client = DevClient(args.host, args.port)
|
||||
client.connect()
|
||||
try:
|
||||
return args.func(client, args)
|
||||
finally:
|
||||
client.close()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Executable
+263
@@ -0,0 +1,263 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Offline verification harness for a built .plugin bundle. Runs without a phone.
|
||||
|
||||
Usage: python3 tools/verify_bundle.py <bundle.plugin>
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import ast
|
||||
import json
|
||||
import sys
|
||||
import traceback
|
||||
import types
|
||||
from dataclasses import dataclass
|
||||
from enum import Enum, auto
|
||||
from pathlib import Path
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parent.parent
|
||||
PLUGIN_JSON = REPO_ROOT / "plugin.json"
|
||||
|
||||
_FAILURES = []
|
||||
|
||||
|
||||
def check(name: str, condition: bool, detail: str = "") -> None:
|
||||
if condition:
|
||||
print(f"PASS {name}")
|
||||
else:
|
||||
message = f"FAIL {name}"
|
||||
if detail:
|
||||
message += f": {detail}"
|
||||
print(message, file=sys.stderr)
|
||||
_FAILURES.append(name)
|
||||
|
||||
|
||||
class HookStrategy(Enum):
|
||||
DEFAULT = auto()
|
||||
MODIFY = auto()
|
||||
CANCEL = auto()
|
||||
MODIFY_FINAL = auto()
|
||||
|
||||
|
||||
@dataclass
|
||||
class HookResult:
|
||||
strategy: HookStrategy = HookStrategy.DEFAULT
|
||||
params: object = None
|
||||
|
||||
|
||||
LOG_LINES: list[str] = []
|
||||
|
||||
|
||||
class BasePlugin:
|
||||
def __init__(self):
|
||||
self._settings = {}
|
||||
|
||||
def get_setting(self, key, default=None):
|
||||
return self._settings.get(key, default)
|
||||
|
||||
def set_setting(self, key, value):
|
||||
self._settings[key] = value
|
||||
|
||||
def add_on_send_message_hook(self):
|
||||
pass
|
||||
|
||||
def on_plugin_load(self):
|
||||
pass
|
||||
|
||||
def on_plugin_unload(self):
|
||||
pass
|
||||
|
||||
def create_settings(self):
|
||||
return []
|
||||
|
||||
|
||||
def fake_log(*args, **kwargs):
|
||||
LOG_LINES.append(" ".join(str(a) for a in args))
|
||||
|
||||
|
||||
class _SettingsRow:
|
||||
def __init__(self, **kwargs):
|
||||
self.__dict__.update(kwargs)
|
||||
|
||||
|
||||
def install_fakes() -> None:
|
||||
base_plugin_mod = types.ModuleType("base_plugin")
|
||||
base_plugin_mod.BasePlugin = BasePlugin
|
||||
base_plugin_mod.HookResult = HookResult
|
||||
base_plugin_mod.HookStrategy = HookStrategy
|
||||
sys.modules["base_plugin"] = base_plugin_mod
|
||||
|
||||
android_utils_mod = types.ModuleType("android_utils")
|
||||
android_utils_mod.log = fake_log
|
||||
sys.modules["android_utils"] = android_utils_mod
|
||||
|
||||
ui_mod = types.ModuleType("ui")
|
||||
ui_mod.__path__ = []
|
||||
sys.modules["ui"] = ui_mod
|
||||
|
||||
ui_settings_mod = types.ModuleType("ui.settings")
|
||||
for cls_name in ("Header", "Input", "Text", "Switch"):
|
||||
setattr(ui_settings_mod, cls_name, type(cls_name, (_SettingsRow,), {}))
|
||||
sys.modules["ui.settings"] = ui_settings_mod
|
||||
ui_mod.settings = ui_settings_mod
|
||||
|
||||
|
||||
class SimpleParams:
|
||||
def __init__(self, message):
|
||||
self.message = message
|
||||
|
||||
|
||||
KEY_TO_DUNDER = {
|
||||
"id": "__id__",
|
||||
"name": "__name__",
|
||||
"description": "__description__",
|
||||
"author": "__author__",
|
||||
"version": "__version__",
|
||||
"icon": "__icon__",
|
||||
"app_version": "__app_version__",
|
||||
"sdk_version": "__sdk_version__",
|
||||
"requirements": "__requirements__",
|
||||
}
|
||||
|
||||
|
||||
def expected_dunders(meta: dict) -> dict:
|
||||
result = {}
|
||||
for key, dunder in KEY_TO_DUNDER.items():
|
||||
if key not in meta:
|
||||
continue
|
||||
value = meta[key]
|
||||
if key in ("description", "icon") and value == "":
|
||||
continue
|
||||
if key == "requirements" and value == []:
|
||||
continue
|
||||
result[dunder] = value
|
||||
return result
|
||||
|
||||
|
||||
def main(argv=None) -> int:
|
||||
argv = sys.argv[1:] if argv is None else argv
|
||||
if len(argv) != 1:
|
||||
print("usage: tools/verify_bundle.py <bundle.plugin>", file=sys.stderr)
|
||||
return 1
|
||||
bundle_path = Path(argv[0])
|
||||
if not bundle_path.is_file():
|
||||
print(f"error: bundle not found at {bundle_path}", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
install_fakes()
|
||||
source = bundle_path.read_text(encoding="utf-8")
|
||||
|
||||
modules_before = set(sys.modules)
|
||||
ns = {"__name__": "hello_bundle"}
|
||||
exec(compile(source, str(bundle_path), "exec"), ns)
|
||||
new_modules = set(sys.modules) - modules_before
|
||||
|
||||
meta = json.loads(PLUGIN_JSON.read_text(encoding="utf-8"))
|
||||
expected = expected_dunders(meta)
|
||||
mismatches = [
|
||||
f"{dunder}: expected {value!r}, got {ns.get(dunder)!r}"
|
||||
for dunder, value in expected.items()
|
||||
if ns.get(dunder) != value
|
||||
]
|
||||
check("metadata constants match plugin.json", not mismatches, "; ".join(mismatches))
|
||||
|
||||
tree = ast.parse(source, filename=str(bundle_path))
|
||||
top_level_assigned = {
|
||||
target.id
|
||||
for node in tree.body
|
||||
if isinstance(node, ast.Assign)
|
||||
for target in node.targets
|
||||
if isinstance(target, ast.Name)
|
||||
}
|
||||
check(
|
||||
"metadata exposed as top-level literal assignments (AST)",
|
||||
set(expected) <= top_level_assigned,
|
||||
f"missing: {set(expected) - top_level_assigned}",
|
||||
)
|
||||
|
||||
plugin_classes = [
|
||||
value
|
||||
for value in ns.values()
|
||||
if isinstance(value, type) and issubclass(value, BasePlugin) and value is not BasePlugin
|
||||
]
|
||||
check(
|
||||
"exactly one BasePlugin subclass defined",
|
||||
len(plugin_classes) == 1,
|
||||
f"found {len(plugin_classes)}",
|
||||
)
|
||||
|
||||
plugin_cls = plugin_classes[0] if len(plugin_classes) == 1 else None
|
||||
if plugin_cls is not None:
|
||||
check(
|
||||
"plugin class __module__ matches executed namespace",
|
||||
plugin_cls.__module__ == ns["__name__"],
|
||||
f"__module__={plugin_cls.__module__!r} != {ns['__name__']!r}",
|
||||
)
|
||||
|
||||
plugin = plugin_cls()
|
||||
LOG_LINES.clear()
|
||||
plugin.on_plugin_load()
|
||||
check(
|
||||
"on_plugin_load emits load log line",
|
||||
any("hello_bundle loaded" in line for line in LOG_LINES),
|
||||
f"log lines: {LOG_LINES}",
|
||||
)
|
||||
|
||||
params = SimpleParams(".hello alice smith")
|
||||
result = plugin.on_send_message_hook(account=0, params=params)
|
||||
check(
|
||||
"'.hello alice smith' renders greeting",
|
||||
params.message == "Hello, Alice Smith!",
|
||||
f"got {params.message!r}",
|
||||
)
|
||||
check(
|
||||
"on_send_message_hook returns MODIFY strategy",
|
||||
result.strategy == HookStrategy.MODIFY,
|
||||
f"got {result.strategy}",
|
||||
)
|
||||
|
||||
plugin.set_setting("template", "{nope}")
|
||||
params2 = SimpleParams(".hello someone")
|
||||
plugin.on_send_message_hook(account=0, params=params2)
|
||||
check(
|
||||
"bad template produces 'Bad template:' message",
|
||||
params2.message.startswith("Bad template:"),
|
||||
f"got {params2.message!r}",
|
||||
)
|
||||
|
||||
# main.py imports `render`/`TemplateError` by name, so they live directly
|
||||
# in the executed namespace — call render() directly to inspect the raised
|
||||
# traceback (proves linecache registration for bundled, non-file modules).
|
||||
render = ns.get("render")
|
||||
template_error = ns.get("TemplateError")
|
||||
traceback_text = ""
|
||||
if render is not None and template_error is not None:
|
||||
try:
|
||||
render("{nope}", "x")
|
||||
except template_error:
|
||||
traceback_text = traceback.format_exc()
|
||||
check(
|
||||
"render() traceback names src/greeting/template.py",
|
||||
"src/greeting/template.py" in traceback_text,
|
||||
f"traceback:\n{traceback_text}",
|
||||
)
|
||||
check(
|
||||
"traceback includes the raising source line text",
|
||||
"unsupported placeholder" in traceback_text,
|
||||
f"traceback:\n{traceback_text}",
|
||||
)
|
||||
|
||||
check(
|
||||
"sys.modules gained no entry for bundled module names",
|
||||
not ({"greeting", "textutil", "main"} & new_modules),
|
||||
f"leaked: {({'greeting', 'textutil', 'main'} & new_modules)}",
|
||||
)
|
||||
|
||||
if _FAILURES:
|
||||
print(f"\n{len(_FAILURES)} check(s) failed: {', '.join(_FAILURES)}", file=sys.stderr)
|
||||
return 1
|
||||
print("\nall checks passed")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in new issue
Block a user