messageLogger/tools/verify_bundle.py

#!/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
    update: object = None


class MenuItemType(Enum):
    MESSAGE_CONTEXT_MENU = auto()


@dataclass
class MenuItemData:
    menu_type: MenuItemType
    text: str
    on_click: object
    item_id: object = None
    icon: object = None
    subtext: object = None
    condition: object = None
    priority: int = 0


class MethodHook:
    pass


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
    base_plugin_mod.MenuItemData = MenuItemData
    base_plugin_mod.MenuItemType = MenuItemType
    base_plugin_mod.MethodHook = MethodHook
    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

    import tempfile

    file_utils_mod = types.ModuleType("file_utils")
    _files = tempfile.mkdtemp(prefix="ml-verify-")
    file_utils_mod.get_files_dir = lambda: _files
    sys.modules["file_utils"] = file_utils_mod

    ui_settings_mod = types.ModuleType("ui.settings")
    for cls_name in ("Header", "Input", "Text", "Switch", "Selector"):
        setattr(ui_settings_mod, cls_name, type(cls_name, (_SettingsRow,), {}))
    sys.modules["ui.settings"] = ui_settings_mod
    ui_mod.settings = ui_settings_mod


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__": "message_logger"}
    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()
        keys = {getattr(row, "key", None) for row in plugin.create_settings()}
        check(
            "settings expose server URL and API token",
            {"server_url", "api_token"} <= keys,
            f"keys: {keys}",
        )

    load = ns.get("_bundle_load")
    check("bundle exposes its module loader", load is not None)
    if load is not None:
        header = load("header")
        text = "✉️ From: Bob\n      ⤷ 💬 Message\n🗑 Deleted at x\n\nHi 😀 there"
        end = header.utf16_len(text)
        body, entities = header.strip(text, [(0, 4, "from"), (end - 5, 5, "bold")])
        check("header stripped from logged copy", body == "Hi 😀 there", repr(body))
        check(
            "entities shifted past the header (UTF-16)",
            entities == [(6, 5, "bold")],
            repr(entities),
        )
        check(
            "text without a header is kept",
            header.strip("plain", [(0, 5, "x")]) == ("plain", [(0, 5, "x")]),
        )

        peers = load("peers")
        check(
            "marked channel id -> channel id",
            peers.channel_id_from_marked(-1000000001234) == 1234,
        )
        chat = load("hooks.chat")
        check(
            "Telethon sender ids -> Android peer ids",
            chat._android_peer_id(-1000000001234) == -1234 and chat._android_peer_id(42) == 42,
        )

        links = load("hooks.links")
        check(
            "private message links parse (with and without topic)",
            links.parse("https://t.me/c/123/45") == (-123, 45)
            and links.parse("t.me/c/123/7/45") == (-123, 45)
            and links.parse("tg://privatepost?channel=123&post=45") == (-123, 45),
        )
        check("non-message links are left alone", links.parse("https://t.me/c/123") is None)

        plan, hour = chat._plan, 3600
        check(
            "sync plan: fresh when unchecked or older than an hour",
            plan(None, 50, 2**31 - 1, 10_000) == (50, 2**31 - 1, True)
            and plan((50, 2**31 - 1, 10_000 - hour), 50, 2**31 - 1, 10_000)
            == (50, 2**31 - 1, True),
        )
        check(
            "sync plan: nothing while the loaded range is checked",
            plan((40, 2**31 - 1, 9_000), 50, 2**31 - 1, 10_000) is None,
        )
        check(
            "sync plan: only the older history loaded since",
            plan((40, 2**31 - 1, 9_000), 10, 2**31 - 1, 10_000) == (10, 39, False)
            and plan((40, 100, 9_000), 50, 120, 10_000) == (101, 120, False),
        )

        store = load("store")
        store.put_server_states(
            0, -5, [{"message_id": i, "edit_count": 1} for i in range(1, 1201)]
        )
        (cached,) = store._conn().execute(
            "SELECT count(*) FROM states WHERE account=0 AND dialog_id=-5"
        ).fetchone()
        check("cache has no per-chat cap", cached == 1200, f"{cached}")
        check(
            "a full check replaces the chat's cached marks",
            store.replace_window(0, -5, 1, 10, [{"message_id": 5, "edit_count": 2}], False, full=True)
            is not None
            and store._conn().execute(
                "SELECT count(*) FROM states WHERE account=0 AND dialog_id=-5"
            ).fetchone()[0]
            == 1,
        )

        store._conn().execute("UPDATE states SET at = at - 40 * 86400")
        store._conn().commit()
        store.put_server_states(0, -6, [{"message_id": 1, "edit_count": 1}])
        removed = store.purge_old(30, force=True)
        left = store._conn().execute("SELECT dialog_id FROM states").fetchall()
        check(
            "age purge drops answers older than the lifetime only",
            removed == 1 and left == [(-6,)],
            f"removed={removed} left={left}",
        )
        check("age purge runs once a day", store.purge_old(1) == 0)
        for name in ("cached_deleted", "coverage", "set_coverage", "expire_coverage", "replace_window"):
            check(f"store.{name} exists", callable(getattr(store, name, None)))

        api = load("api")
        client = api.Api(BasePlugin())
        check("API unconfigured without URL and token", not client.configured)
        client.plugin.set_setting("server_url", "https://panel.example/ ")
        client.plugin.set_setting("api_token", "tlp_x, tlp_y\ntlp_z")
        check("several tokens, any separator", client.tokens() == ["tlp_x", "tlp_y", "tlp_z"])
        check(
            "API base URL",
            client._base() == "https://panel.example/api/v1/client/message-logger",
            client._base(),
        )

    leaked = {"main", "header", "api", "store", "peers", "hooks", "views"} & new_modules
    check("sys.modules gained no entry for bundled module names", 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())