Initial commit

This commit is contained in:
exteraGramPlugins committed 2026-09-23 09:10:22 +00:00
commit e0586a3423
31 files changed
+2335

No files matched your search

+345
View File
@@ -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())
+219
View File
@@ -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())
+220
View File
@@ -0,0 +1,220 @@
#!/usr/bin/env python3
"""Publishes a built .plugin file to a Gitea release.
Creates the release for the given tag if it does not exist, replaces any
same-named asset already attached, then uploads the artifact. Re-running is
safe: the release is reused and the asset is replaced rather than duplicated.
Stdlib only, so it runs on any runner image without a pip install step.
Usage:
tools/release.py --api <repo-api-url> --tag v1.0.0 --artifact dist/x.plugin
The token is read from RELEASE_TOKEN by default; override with --token-env.
Deliberately not GITEA_TOKEN: the Actions runner injects its own value for
that name into every step, which overrides a step-level `env:` entry.
"""
from __future__ import annotations
import argparse
import json
import mimetypes
import os
import sys
import urllib.error
import urllib.parse
import urllib.request
import uuid
from pathlib import Path
from typing import Optional, Tuple
REPO_ROOT = Path(__file__).resolve().parent.parent
def fail(message: str) -> "None":
print(f"error: {message}", file=sys.stderr)
raise SystemExit(1)
def request(
method: str,
url: str,
token: str,
body: Optional[bytes] = None,
content_type: Optional[str] = None,
) -> Tuple[int, bytes]:
"""Performs one API call, returning (status, body) instead of raising."""
req = urllib.request.Request(url, data=body, method=method)
req.add_header("Authorization", f"token {token}")
req.add_header("Accept", "application/json")
if content_type:
req.add_header("Content-Type", content_type)
try:
with urllib.request.urlopen(req) as response:
return response.status, response.read()
except urllib.error.HTTPError as exc:
return exc.code, exc.read()
except urllib.error.URLError as exc:
fail(f"{method} {url} failed: {exc.reason}")
raise # unreachable, keeps type checkers quiet
def find_release(api: str, tag: str, token: str) -> Optional[dict]:
status, payload = request("GET", f"{api}/releases/tags/{tag}", token)
if status == 404:
return None
if status != 200:
fail(f"looking up release {tag} returned HTTP {status}: {payload[:400]!r}")
return json.loads(payload)
def create_release(api: str, tag: str, token: str, commitish: str, notes: str) -> dict:
"""Creates the release, which also creates the tag if it does not exist."""
body = json.dumps(
{
"tag_name": tag,
"target_commitish": commitish,
"name": tag,
"body": notes,
"draft": False,
"prerelease": False,
}
).encode()
status, payload = request("POST", f"{api}/releases", token, body, "application/json")
if status != 201:
fail(f"creating release {tag} returned HTTP {status}: {payload[:400]!r}")
return json.loads(payload)
def update_release(api: str, release_id: int, token: str, commitish: str, notes: str) -> dict:
"""Refreshes an existing release so a rebuild reports the commit it came from.
target_commitish is deliberately not changed: the tag already points at the
commit that first introduced this version, and moving it would silently
rewrite history for anyone who already fetched it.
"""
body = json.dumps({"body": notes}).encode()
status, payload = request(
"PATCH", f"{api}/releases/{release_id}", token, body, "application/json"
)
if status != 200:
fail(f"updating release {release_id} returned HTTP {status}: {payload[:400]!r}")
return json.loads(payload)
def delete_existing_assets(api: str, release_id: int, name: str, token: str) -> int:
"""Removes same-named assets, since Gitea keeps duplicates side by side."""
status, payload = request("GET", f"{api}/releases/{release_id}/assets", token)
if status != 200:
fail(f"listing assets returned HTTP {status}: {payload[:400]!r}")
removed = 0
for asset in json.loads(payload) or []:
if asset.get("name") != name:
continue
status, payload = request(
"DELETE", f"{api}/releases/{release_id}/assets/{asset['id']}", token
)
if status not in (204, 404):
fail(f"deleting asset {asset['id']} returned HTTP {status}: {payload[:400]!r}")
removed += 1
return removed
def upload_asset(api: str, release_id: int, path: Path, name: str, token: str) -> dict:
"""Uploads the artifact as multipart/form-data."""
boundary = uuid.uuid4().hex
mime = mimetypes.guess_type(name)[0] or "application/octet-stream"
payload = b"".join(
[
f"--{boundary}\r\n".encode(),
f'Content-Disposition: form-data; name="attachment"; filename="{name}"\r\n'.encode(),
f"Content-Type: {mime}\r\n\r\n".encode(),
path.read_bytes(),
f"\r\n--{boundary}--\r\n".encode(),
]
)
url = f"{api}/releases/{release_id}/assets?name={urllib.parse.quote(name)}"
status, response = request(
"POST", url, token, payload, f"multipart/form-data; boundary={boundary}"
)
if status != 201:
fail(f"uploading {name} returned HTTP {status}: {response[:400]!r}")
return json.loads(response)
def main(argv: Optional[list] = None) -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--api", required=True, help="repo API base, e.g. https://host/api/v1/repos/owner/repo")
parser.add_argument("--tag", required=True, help="release tag")
parser.add_argument("--artifact", required=True, help="file to attach")
parser.add_argument("--commitish", default="", help="target commit for a new tag")
parser.add_argument("--asset-name", default="", help="attachment name; defaults to the file name")
parser.add_argument(
"--token-env",
default="RELEASE_TOKEN",
help="preferred environment variable holding the API token",
)
args = parser.parse_args(argv)
# Preferred name first, then the runner-provided job token. GITEA_TOKEN is
# not used as the primary name because the Actions runner injects its own
# value for it into every step, which overrides a step-level `env:` entry
# of the same name -- but it is a valid last resort, since it can publish
# releases in its own repository.
candidates = [args.token_env, "RELEASE_TOKEN", "GITEA_TOKEN", "GITHUB_TOKEN"]
token, token_source = "", ""
for name in candidates:
value = os.environ.get(name, "").strip()
if value:
token, token_source = value, name
break
if not token:
present = sorted(n for n in candidates if n in os.environ)
fail(
f"no API token found; looked at {', '.join(dict.fromkeys(candidates))}. "
+ (
f"These exist but are empty: {', '.join(present)}. "
"An organisation secret is only exposed to a repository when Actions "
"is enabled on that repository, so check Repo > Settings > Actions."
if present
else "None of them are set in the environment."
)
)
print(f"using token from {token_source}")
artifact = Path(args.artifact)
if not artifact.is_file() or artifact.stat().st_size == 0:
fail(f"artifact {artifact} is missing or empty")
asset_name = args.asset_name or artifact.name
api = args.api.rstrip("/")
version = json.loads((REPO_ROOT / "plugin.json").read_text(encoding="utf-8"))["version"]
notes = [f"Automated build of version {version}."]
if args.commitish:
notes.append(f"Built from `{args.commitish[:12]}`.")
notes.append("Install by opening the attached `.plugin` file in exteraGram.")
notes_text = "\n\n".join(notes)
release = find_release(api, args.tag, token)
if release is None:
release = create_release(api, args.tag, token, args.commitish, notes_text)
print(f"created release {args.tag}")
else:
# Version unchanged since the last push: keep the same tag and refresh
# it in place so the attachment always matches the newest commit.
release = update_release(api, int(release["id"]), token, args.commitish, notes_text)
print(f"rebuilding existing release {args.tag}")
release_id = int(release["id"])
removed = delete_existing_assets(api, release_id, asset_name, token)
if removed:
print(f"removed {removed} previous copy/copies of {asset_name}")
asset = upload_asset(api, release_id, artifact, asset_name, token)
print(f"uploaded {asset_name} ({artifact.stat().st_size} bytes) -> {asset.get('browser_download_url', '')}")
return 0
if __name__ == "__main__":
sys.exit(main())
+263
View File
@@ -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())