cthulhu-tentacles/tools/check.py

# Copyright (c) 2026 Storm Dragon
# SPDX-License-Identifier: GPL-3.0-or-later
"""Check the repository and the built Tentacle without running a reader."""

import json
import hashlib
import os
from pathlib import Path
import select
import subprocess
import tomllib
from urllib.parse import urlparse


def require(condition, message):
    if not condition:
        raise RuntimeError(message)


def read_toml(path):
    with path.open("rb") as source:
        return tomllib.load(source)


def check_metadata(repoRoot):
    repository = read_toml(repoRoot / "repository.toml")
    catalog = read_toml(repoRoot / "catalog.toml")
    require(repository["format-version"] == catalog["format-version"] == 1,
            "Unsupported metadata format")
    require(repository["id"] == "storm.cthulhu-tentacles" and repository["name"],
            "Repository identity is missing")
    require([item["id"] for item in catalog["tentacles"]] == ["hello-world", "wormhole"],
            "Catalog must list Hello World and Wormhole")
    for item in catalog["tentacles"]:
        name = item["id"]
        require(item["manifest"] == f"{name}/tentacle.toml", "Manifest path must match ID")
        manifest = read_toml(repoRoot / item["manifest"])
        require(manifest["format-version"] == 1 and manifest["id"] == name
                and manifest["name"] and manifest["version"] == ("0.1.2" if name == "wormhole" else "0.1.0"), "Invalid Tentacle metadata")
        expectedGrants = ["presentation"] + (["interaction"] if name == "wormhole" else [])
        expectedProgram = "wormhole-tentacle" if name == "wormhole" else name
        require(manifest["entry-point"] == [f"bin/{expectedProgram}"]
                and manifest["capabilities"] == expectedGrants
                and manifest["apis"] == ["commands"] + expectedGrants,
                "Unexpected launch or grants")
        require((repoRoot / "LICENSE").is_file() and (repoRoot / name / "LICENSE").is_file()
                and (repoRoot / name / "README.md").is_file(), "License and help are required")
        seen = set()
        for release in item.get("releases", []):
            key = (release["platform"], release["architecture"])
            require(key not in seen and key[0] == "linux" and key[1] in ("amd64", "arm64"),
                    "Invalid or duplicate release target")
            seen.add(key)
            parsed = urlparse(release["url"])
            require(parsed.scheme == "https" and parsed.hostname and not parsed.username
                    and not parsed.fragment, "Release URL must be HTTPS without credentials")
            digest = release["sha256"]
            require(len(digest) == 64 and all(character in "0123456789abcdef" for character in digest),
                    "Invalid SHA-256 digest")
            artifactPath = repoRoot / "dist" / parsed.path.rsplit("/", 1)[-1]
            if artifactPath.exists():
                require(hashlib.sha256(artifactPath.read_bytes()).hexdigest() == digest,
                        "Prepared package does not match its catalog digest")


def check_registration(programPath, name):
    apis = ["commands", "presentation"] + (["interaction"] if name == "wormhole" else [])
    with subprocess.Popen([str(programPath)], stdin=subprocess.PIPE,
                          stdout=subprocess.PIPE, stderr=subprocess.PIPE) as process:
        try:
            initialization = {
                "jsonrpc": "2.0", "id": "1", "method": "initialize",
                "params": {"protocol": {"major": 1, "minor": 0}, "session": "check",
                           "apis": apis,
                           "capabilities": apis},
            }
            process.stdin.write((json.dumps(initialization) + "\n").encode())
            process.stdin.flush()
            require(select.select([process.stdout], [], [], 5)[0], "Registration timed out")
            registration = json.loads(process.stdout.readline())
            require(registration["id"] == "1" and registration["result"]["session"] == "check",
                    "Invalid registration response")
            commands = registration["result"]["commands"]
            require(len(commands) == (9 if name == "wormhole" else 1) and commands[0]["id"] == name
                    and commands[0]["presentation"] is True, "Unexpected commands or grant")
            require(commands[0]["bindings"] == [
                {"layout": layout, "scope": "global", "key": ("Cthulhu+Ctrl+Shift+W" if name == "wormhole" else "Cthulhu+Alt+Shift+H")}
                for layout in ("desktop", "laptop")], "Wrong suggested shortcut")
            if name == "wormhole":
                require([command["id"] for command in commands if command.get("settings")]
                        == ["settings"], "Wrong registered settings entry point")
            process.stdin.close()
            require(process.wait(timeout=5) == 0, "Tentacle did not stop cleanly on EOF")
        finally:
            if process.poll() is None:
                process.kill()
                process.wait(timeout=5)


def check_host(repoRoot, programPath, name):
    readerRoot = repoRoot.parent / "cthulhu"
    environment = os.environ.copy()
    environment.setdefault("GOCACHE", "/tmp/cthulhu-cache")
    flags = ["-interaction"] if name == "wormhole" else []
    result = subprocess.run([
        "go", "run", "./cmd/tentacle-host", "-presentation",
        *flags, "-command", name, str(programPath),
    ], cwd=readerRoot, env=environment, capture_output=True, text=True, timeout=60, check=True)
    if name == "wormhole":
        value = json.loads(result.stdout)
        require(value.get("form", {}).get("title") == "Wormhole menu"
                or value == {"speech": "Magic wormhole is not installed",
                             "braille": "Magic wormhole is not installed", "uninterruptible": True},
                "Host returned an unexpected Wormhole menu")
        result = subprocess.run([
            "go", "run", "./cmd/tentacle-host", "-presentation", "-interaction",
            "-command", "settings", str(programPath),
        ], cwd=readerRoot, env=environment, capture_output=True, text=True,
            timeout=60, check=True)
        value = json.loads(result.stdout)
        require(value.get("form", {}).get("title") == "Wormhole preferences"
                and value.get("next") == "save-settings",
                "Host returned an unexpected Wormhole settings form")
        return
    require(json.loads(result.stdout) == {
        "speech": "Hello world!", "braille": "Hello world!", "uninterruptible": True,
    }, "The actual host returned an unexpected greeting")


def main():
    repoRoot = Path(__file__).resolve().parents[1]
    programPath = repoRoot / "hello-world/bin/hello-world"
    check_metadata(repoRoot)
    for name, binary in (("hello-world", "hello-world"), ("wormhole", "wormhole-tentacle")):
        programPath = repoRoot / name / "bin" / binary
        check_registration(programPath, name)
        check_host(repoRoot, programPath, name)
    print("Repository metadata, shortcut/settings registration and actual-host greeting/menu/settings checks passed.")


if __name__ == "__main__":
    main()