clientCommands/tools/verify_bundle.py

#!/usr/bin/env python3
"""Offline verification harness for a built .plugin bundle. Runs without a phone.

Executes the bundle against faked SDK modules and checks what can be checked
off the device: metadata, the argument parser (in step with telegram-panel's),
HTML responses, the library API other plugins use, the command catalog
against a fake panel, and plugin commands run by the runner.

Usage: python3 tools/verify_bundle.py <bundle.plugin>
"""
from __future__ import annotations

import ast
import json
import sys
import tempfile
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)


def raises(fn, error_type, text=None):
    try:
        fn()
    except error_type as exc:
        return text is None or text in str(exc)
    return False


class HookStrategy(Enum):
    DEFAULT = auto()
    MODIFY = auto()
    CANCEL = auto()
    MODIFY_FINAL = auto()


@dataclass
class HookResult:
    strategy: HookStrategy = HookStrategy.DEFAULT
    params: object = None


class MethodHook:
    pass


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, reload_settings=False):
        self._settings[key] = value

    def add_on_send_message_hook(self):
        pass


class _SettingsRow:
    def __init__(self, **kwargs):
        self.__dict__.update(kwargs)


class _UserConfig:
    """org.telegram.messenger.UserConfig: account slot n is Telegram user 1000 + n."""

    def __init__(self, slot):
        self.slot = slot

    @classmethod
    def getInstance(cls, slot):
        return cls(slot)

    def getClientUserId(self):
        return 1000 + self.slot


FILES_DIR = tempfile.mkdtemp(prefix="cc-verify-")


def install_fakes() -> None:
    def module(name, **attrs):
        mod = types.ModuleType(name)
        mod.__dict__.update(attrs)
        sys.modules[name] = mod
        return mod

    module(
        "base_plugin",
        BasePlugin=BasePlugin,
        HookResult=HookResult,
        HookStrategy=HookStrategy,
        MethodHook=MethodHook,
    )
    module("android_utils", log=lambda *a, **k: None, run_on_ui_thread=lambda fn, delay=0: fn())
    module("file_utils", get_files_dir=lambda: FILES_DIR)
    ui = module("ui")
    ui.__path__ = []
    ui.settings = module(
        "ui.settings",
        **{n: type(n, (_SettingsRow,), {}) for n in ("Header", "Input", "Text", "Switch", "Divider")},
    )
    for name in ("org", "org.telegram"):
        module(name).__path__ = []
    module("org.telegram.messenger", UserConfig=_UserConfig)


class FakeApi:
    """The panel, as api.Api's methods answer."""

    def __init__(self, tokens):
        self._tokens = tokens  # token -> (tg id, [command json]) or an ApiError
        self.configured = True

    def tokens(self):
        return list(self._tokens)

    def _of(self, token):
        found = self._tokens[token]
        if isinstance(found, Exception):
            raise found
        return found

    def me(self, token):
        tg_id, _ = self._of(token)
        return {"tg_id": tg_id, "account_name": f"acc{tg_id}", "running": True}

    def commands(self, token):
        _, commands = self._of(token)
        return {"commands": commands, "refresh_seconds": 3600}


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 check_metadata(source, ns, bundle_path):
    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))
    assigned = {
        t.id for node in tree.body if isinstance(node, ast.Assign) for t in node.targets if isinstance(t, ast.Name)
    }
    check("metadata exposed as top-level literal assignments (AST)", set(expected) <= assigned)
    plugins = [v for v in ns.values() if isinstance(v, type) and issubclass(v, BasePlugin) and v is not BasePlugin]
    check("exactly one BasePlugin subclass defined", len(plugins) == 1, f"found {len(plugins)}")
    return plugins[0] if len(plugins) == 1 else None


def check_parser(load):
    model, parse = load("commands.model"), load("commands.parse")
    Param, Command = model.Param, model.Command
    ban = Command(
        "ban",
        [
            Param("days", "integer"),
            Param("mode", "choice", False, default="soft", choices=("soft", "hard")),
            Param("silent", "boolean", False, default=False),
            Param("reason", "string", False),
        ],
        "Bans the replied user.",
        aliases=["b"],
        reply="required",
    )
    # The same cases as telegram-panel's tests/test_client_commands.py.
    check("split /Ban@me", parse.split_command("/Ban@me 3") == ("ban", "3"))
    check("no command mid-text", parse.split_command("hello /ban") is None)
    check(
        "positional, choice, boolean and the greedy last string",
        parse.parse(ban, "/b 3 HARD yes for spamming links")
        == {"days": 3, "mode": "hard", "silent": True, "reason": "for spamming links"},
    )
    check(
        "named and quoted",
        parse.parse(ban, '/ban reason:"too loud" days:7') == {"days": 7, "mode": "soft", "silent": False, "reason": "too loud"},
    )
    check("an empty name: is not given", parse.parse(ban, "/ban 1 mode:")["mode"] == "soft")
    for text, error in (
        ("/ban", "Missing required parameter: days"),
        ("/ban x", "days must be a whole number"),
        ("/ban 1 medium", "mode must be one of: soft, hard"),
        ("/ban 1 soft maybe", "silent must be yes or no"),
        ('/ban 1 "open', "Unclosed quote"),
        ("/ban days:1 days:2", "days is given twice"),
    ):
        check(f"error for {text!r}", raises(lambda: parse.parse(ban, text), parse.CommandError, error))
    add = Command("add", [Param("a", "integer"), Param("b", "integer")])
    check("extra words without a string to take them", raises(lambda: parse.parse(add, "/add 1 2 3"), parse.CommandError, "Unexpected arguments: 3"))

    check("bare /prefix is a name being typed", parse.command_name_at("/ba") == "ba" and parse.command_name_at("/ba ") is None)
    typing = parse.analyze(ban, "/ban 3 ", None, has_reply=True)
    check("analyze: next free parameter is current", typing.current == "mode" and typing.complete, typing.current)
    named = parse.analyze(ban, "/ban reason:so", None, has_reply=True)
    check("analyze: a named parameter at the cursor", named.current == "reason" and named.missing == ["days"])
    check("analyze: missing days blocks running", not named.complete and "days" in named.error)
    no_reply = parse.analyze(ban, "/ban 3", None, has_reply=False)
    check("analyze: a required reply blocks running", no_reply.error == "Reply to a message to use /ban")
    middle = parse.analyze(ban, "/ban 3 hard", 6, has_reply=True)
    check("analyze: the cursor inside a word", middle.current == "days", middle.current)
    check("usage and hint", ban.usage() == "/ban <days> [mode] [silent] [reason]" and ban.hint().startswith("<days> [mode]"))
    back = Command.from_json(ban.to_json())
    check("command JSON round trip", back.to_json() == ban.to_json() and back.names == ("ban", "b"))
    check("bad names are refused", raises(lambda: Command("no spaces"), ValueError))


def check_markup(load):
    markup = load("commands.markup")
    text, entities = markup.parse('😀 <b>bold</b> <a href="https://x.y">link</a><br><tg-spoiler>s</tg-spoiler>')
    check("HTML text", text == "😀 bold link\ns", repr(text))
    check(
        "HTML entities in UTF-16 units",
        [(e["type"], e["offset"], e["length"]) for e in entities] == [("bold", 3, 4), ("text_link", 8, 4), ("spoiler", 13, 1)],
        repr(entities),
    )
    _, pre = markup.parse('<pre><code class="language-py">x = 1</code></pre>')
    check("pre with a language", pre == [{"type": "pre", "offset": 0, "length": 5, "language": "py"}], repr(pre))
    _, mention = markup.parse('<a href="tg://user?id=42">Ann</a>')
    check("tg://user links are mentions", mention[0]["type"] == "mention_name" and mention[0]["user_id"] == 42)
    check("escape", markup.escape("<a&b>") == "&lt;a&amp;b&gt;")


def check_library(load):
    import builtins

    sys.modules.pop("client_commands", None)
    builtins.__dict__.pop("__client_commands_state__", None)
    queued = []
    early = types.ModuleType("client_commands")
    early.pending = [lambda api: queued.append(api.register("early", lambda ctx: "hi", owner="other"))]
    sys.modules["client_commands"] = early
    lib = load("library").Library()
    lib.install()
    api = sys.modules["client_commands"]
    check("library published as client_commands", api is early and api.available and hasattr(api, "register"))
    check("setups queued before loading ran", len(queued) == 1 and "pending" not in api.__dict__)
    api.register(
        "roll",
        lambda ctx, sides=6: f"🎲 <b>{sides}</b>",
        params=[api.Param("sides", "integer", required=False, default=6)],
        summary="Rolls a die",
        owner="dice",
    )
    names = [c["name"] for c in api.commands()]
    check("registered commands listed", names == ["early", "roll"], repr(names))
    lib.uninstall()
    sys.modules.pop("client_commands")  # the engine drops modules a plugin added
    again = load("library").Library()
    again.install()
    check("registrations survive a reload", [c.name for c in again.local_commands()] == ["early", "roll"])
    api.register("late", lambda ctx: None, owner="other")  # an api kept from before
    check("an old api registers into the same registry", "late" in [c.name for c in again.local_commands()])
    api = sys.modules["client_commands"]
    check("unregister_owner", api.unregister_owner("other") == 2 and not api.unregister("early"))
    return again


def check_catalog(load, library):
    from_api = load("api")
    catalog_mod = load("catalog")
    getid = {
        "name": "getid",
        "summary": "Ids",
        "reply": "optional",
        "params": [{"name": "target", "type": "string", "required": False}],
    }
    roll_shadow = {"name": "roll", "summary": "Panel roll", "params": []}
    api = FakeApi(
        {
            "tok_a": (1000, [getid, roll_shadow]),
            "tok_b": (1001, [{"name": "ping"}]),
            "tok_bad": from_api.ApiError("403: Not a commands token", 403),
        }
    )
    plugin = BasePlugin()
    catalog = catalog_mod.Catalog(plugin, api, library)
    catalog._refresh_due()
    names0 = [c.name for c in catalog.commands_for(0)]
    check("account 0: panel commands, then plugins'", names0 == ["getid", "roll"], repr(names0))
    check("a panel command shadows a plugin's", catalog.find(0, "roll").summary == "Panel roll")
    names1 = [c.name for c in catalog.commands_for(1)]
    check("account 1 has its own panel commands", names1 == ["ping", "roll"], repr(names1))
    check("tokens picked by Telegram id", catalog.token_for(0) == "tok_a" and catalog.token_for(1) == "tok_b")
    check("a refused token is reported", "tok_bad" in catalog.errors())
    check("accounts without a token only get plugins'", [c.name for c in catalog.commands_for(5)] == ["roll"])
    cached = catalog_mod.Catalog(plugin, FakeApi({}), library)
    check("commands cached on disk", [c.name for c in cached.commands_for(0)] == ["getid", "roll"])
    plugin.set_setting("server_commands", False)
    check("panel commands can be turned off", [c.name for c in catalog.commands_for(0)] == ["roll"])
    plugin.set_setting("server_commands", True)
    api._tokens.pop("tok_b")
    api._tokens.pop("tok_bad")
    catalog._refresh_due()
    check("a removed token's commands go", [c.name for c in catalog.commands_for(1)] == ["roll"])


def check_runner(load, library):
    runner_mod = load("runner")
    runner = runner_mod.Runner(BasePlugin(), None, None, library)
    roll = next(c for c in library.local_commands() if c.name == "roll")
    inv = types.SimpleNamespace(
        account=0, dialog_id=5, topic_id=0, reply=None, text="/roll 20", command=roll, send_to_chat=False
    )
    result = runner._local(inv)
    check(
        "plugin command response parsed",
        result["ok"] and result["text"] == "🎲 20" and result["entities"] == [{"type": "bold", "offset": 3, "length": 2}],
        repr(result),
    )
    api = sys.modules["client_commands"]

    def strict(ctx, n):
        raise api.CommandError(f"no {n}")

    api.register("strict", strict, params=[{"name": "n", "type": "integer", "required": True}], owner="t")
    inv.command = next(c for c in library.local_commands() if c.name == "strict")
    inv.text = "/strict 3"
    check("CommandError is a user-facing error", runner._local(inv) == {"ok": False, "error": "no 3"})
    inv.text = "/strict"
    check("bad arguments are a user-facing error", runner._local(inv)["error"] == "Missing required parameter: n")
    runner.bypass("/strict 1")
    check("bypassed text goes through once", runner.bypassed("/strict 1", consume=True) and not runner.bypassed("/strict 1"))


def check_send_hook(plugin_cls):
    plugin = plugin_cls()
    plugin.runner = types.SimpleNamespace(bypassed=lambda text, consume=False: False)
    plugin.catalog = types.SimpleNamespace(find=lambda account, name: None)
    params = types.SimpleNamespace(message="hello", scheduleDate=0)
    check("plain messages pass the send hook", plugin.on_send_message_hook(0, params).strategy == HookStrategy.DEFAULT)
    params.message = "/unknown"
    check("unknown commands pass the send hook", plugin.on_send_message_hook(0, params).strategy == HookStrategy.DEFAULT)
    model = sys.modules.get("_cc_model")
    demo = model.Command("demo", [model.Param("count", "integer")])
    refills = []
    plugin.catalog = types.SimpleNamespace(find=lambda account, name: demo if name == "demo" else None)
    plugin._refill = lambda *args: refills.append(args)
    params = types.SimpleNamespace(message="/demo", scheduleDate=0, peer=5, replyToMsg=None, replyToTopMsg=None)
    result = plugin.on_send_message_hook(0, params)
    check(
        "an incomplete command is never sent: it goes back into the field",
        result.strategy == HookStrategy.CANCEL and refills and refills[0][2] == "/demo",
        repr(refills),
    )
    keys = {getattr(r, "key", None) for r in plugin.create_settings()}
    check("settings have the panel URL and tokens", {"server_url", "api_tokens", "send_to_chat"} <= keys, repr(keys))


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__": "client_commands_plugin"}
    exec(compile(source, str(bundle_path), "exec"), ns)

    plugin_cls = check_metadata(source, ns, bundle_path)
    load = ns.get("_bundle_load")
    check("bundle exposes its module loader", load is not None)
    if load is not None:
        sys.modules["_cc_model"] = load("commands.model")
        check_parser(load)
        check_markup(load)
        library = check_library(load)
        check_catalog(load, library)
        check_runner(load, library)
    if plugin_cls is not None:
        check_send_hook(plugin_cls)

    sys.modules.pop("_cc_model", None)
    leaked = {"commands", "hooks", "catalog", "runner", "main"} & (set(sys.modules) - modules_before)
    check("bundled modules stay out of sys.modules", not leaked, f"leaked: {leaked}")

    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())