diff options
| author | Andrew Branson <andrew.branson@jolla.com> | 2026-05-15 12:21:39 +0200 |
|---|---|---|
| committer | Andrew Branson <andrew.branson@jolla.com> | 2026-05-15 12:21:39 +0200 |
| commit | 1f14c5483ee111105f94d66fb3b82208946d914a (patch) | |
| tree | d931715d7a3335324fbcdc51fcef1e0cef590235 | |
Initial Sailfish devel MCP
| -rw-r--r-- | .gitignore | 9 | ||||
| -rw-r--r-- | LICENSE | 15 | ||||
| -rw-r--r-- | README.md | 150 | ||||
| -rwxr-xr-x | bin/sailfish-devel-mcp | 8 | ||||
| -rw-r--r-- | examples/config.json | 19 | ||||
| -rw-r--r-- | pyproject.toml | 20 | ||||
| -rw-r--r-- | src/sailfish_devel_mcp/__init__.py | 4 | ||||
| -rw-r--r-- | src/sailfish_devel_mcp/__main__.py | 6 | ||||
| -rw-r--r-- | src/sailfish_devel_mcp/config.py | 178 | ||||
| -rw-r--r-- | src/sailfish_devel_mcp/runner.py | 115 | ||||
| -rw-r--r-- | src/sailfish_devel_mcp/server.py | 196 | ||||
| -rw-r--r-- | src/sailfish_devel_mcp/tools.py | 1282 | ||||
| -rw-r--r-- | src/sailfish_devel_mcp/vendor/__init__.py | 2 | ||||
| -rwxr-xr-x | src/sailfish_devel_mcp/vendor/build_sailfishos.py | 1459 | ||||
| -rw-r--r-- | tests/test_server.py | 418 |
15 files changed, 3881 insertions, 0 deletions
diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..3bde52c --- /dev/null +++ b/.gitignore @@ -0,0 +1,9 @@ +.mypy_cache/ +.pytest_cache/ +__pycache__/ +*.egg-info/ +*.pyc +.venv/ +build/ +dist/ + @@ -0,0 +1,15 @@ +0BSD + +Copyright (C) 2026 Andrew Branson + +Permission to use, copy, modify, and/or distribute this software for any +purpose with or without fee is hereby granted. + +THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH +REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY AND +FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, +INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM LOSS +OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER +TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF +THIS SOFTWARE. + diff --git a/README.md b/README.md new file mode 100644 index 0000000..54e9247 --- /dev/null +++ b/README.md @@ -0,0 +1,150 @@ +# Sailfish Devel MCP + +Host-side Model Context Protocol server for Sailfish OS development workflows. + +This first iteration is dependency-free: it implements MCP over stdio directly +with Python's standard library and exposes typed tools around commands that are +easy to get wrong during day-to-day Sailfish work. + +## Scope + +The server currently exposes tools for: + +- Sailfish device access over SSH +- defaultuser session-bus calls +- Lipstick screenshots +- touchscreen discovery and tap/swipe injection +- topmost window PID lookup +- process map inspection +- journal log reads +- RPM copy/install on a device +- system and user service management +- Docker/mb2 RPM builds through the local `build-sailfishos` helper +- Jolla OBS result and build-log lookup through `osc` +- repository status and search under the configured git root +- RPM spec metadata summaries +- QML translation search and Sailfish ternary translation checks + +The committed defaults are deliberately generic. Device tools default to the +placeholder SSH target `root@device`, the Sailfish user-session bus at +`/run/user/100000/dbus/user_bus_socket`, `~/git` as the local source root, and +the vendored build helper at +`src/sailfish_devel_mcp/vendor/build_sailfishos.py`. Put a config file at +`~/.config/sailfish-devel-mcp/config.json` or pass `--config` to provide your +real device and OBS settings. Device entries can also carry the preferred +user, architecture, and current release label. If an installed SDK is +available, set `paths.local_sdk` to its `sdk-chroot` path; builds will use it +when it has a target matching the requested release and architecture, otherwise +they fall back to the coderus SDK image. + +## Running + +From a checkout: + +```sh +/path/to/sailfish-devel-mcp/bin/sailfish-devel-mcp +``` + +To inspect the effective configuration: + +```sh +/path/to/sailfish-devel-mcp/bin/sailfish-devel-mcp --dump-config +``` + +Example MCP client configuration: + +```json +{ + "mcpServers": { + "sailfish-devel": { + "command": "/path/to/sailfish-devel-mcp/bin/sailfish-devel-mcp" + } + } +} +``` + +## Configuration + +Example: + +```json +{ + "default_device": "phone", + "devices": { + "phone": { + "ssh_target": "root@phone", + "username": "defaultuser", + "architecture": "aarch64", + "release": "live", + "user_bus_runtime_dir": "/run/user/100000", + "user_bus_address": "unix:path=/run/user/100000/dbus/user_bus_socket" + } + }, + "paths": { + "git_root": "/home/you/git", + "ssh_config": "/home/you/.ssh/config", + "local_sdk": "/srv/mer/sdks/sfossdk/sdk-chroot", + "osc_api_alias": "your-obs-alias" + } +} +``` + +## Tool Notes + +The server keeps local paths scoped to the configured git root and `/tmp` for +tools that read or write files. Device access still uses SSH, so the usual SSH +prompts, permissions, and command failures are surfaced as tool results. + +Mutating tools are annotated as non-read-only: + +- `sailfish_device_lipstick_screenshot` +- `sailfish_device_touch` +- `sailfish_device_user_bus_call` +- `sailfish_device_install_rpm` +- `sailfish_device_restart_service` +- `sailfish_build_rpm` + +`sailfish_device_lipstick_screenshot` defaults to +`~/Pictures/Screenshots/lipstick-<timestamp>.png` under `/home/<username>`, +where `username` comes from the device config. Lipstick rejects screenshot save +paths outside the home directory. When that directory is missing, the tool +derives ownership with `stat -L`, creates `Pictures` as that owner/group with +mode `775`, and creates `Pictures/Screenshots` as that owner and the +`privileged` group with mode `755` when the group exists. + +`sailfish_device_touch` supports `discover`, `tap`, and `swipe`. Discovery +prints `/proc/bus/input/devices` and can also run `evdev_trace -i` when +`include_evdev_trace` is true. Tap and swipe auto-select a likely touchscreen +input event device unless `input_device` is supplied. Coordinates are raw +input/display coordinates, so pair this tool with a current screenshot when +choosing points. + +`sailfish_build_rpm` can use a configured device's `architecture` and `release` +as defaults when the call includes `device`. If `paths.local_sdk` is set, the +build helper first checks the installed SDK targets. `live` uses the unversioned +local target for the requested architecture, for example `aarch64`; a named +release such as `5.0.0` uses a matching versioned local target such as +`aarch64-5.0.0`. If the required target is not installed, the helper falls back +to the release-specific coderus Docker SDK image. The wrapper image defaults to +`sailfish-sdk-build-engine:$USER` and can be overridden with +`SAILFISH_SDK_BUILD_ENGINE_IMAGE`. + +Read-only tools include the journal, topmost PID, process maps, OBS lookup, +repo search, spec summary, and QML checks. + +## Smoke Test + +```sh +printf '%s\n' \ + '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-06-18","capabilities":{},"clientInfo":{"name":"smoke","version":"0"}}}' \ + '{"jsonrpc":"2.0","id":2,"method":"tools/list","params":{}}' \ + | /path/to/sailfish-devel-mcp/bin/sailfish-devel-mcp +``` + +## Development + +Run tests without installing the package: + +```sh +PYTHONPATH=src python3 -m unittest discover -s tests +``` diff --git a/bin/sailfish-devel-mcp b/bin/sailfish-devel-mcp new file mode 100755 index 0000000..5105b01 --- /dev/null +++ b/bin/sailfish-devel-mcp @@ -0,0 +1,8 @@ +#!/bin/sh +set -eu + +root=$(CDPATH= cd -- "$(dirname -- "$0")/.." && pwd) +PYTHONPATH="$root/src${PYTHONPATH:+:$PYTHONPATH}" +export PYTHONPATH +exec python3 -m sailfish_devel_mcp "$@" + diff --git a/examples/config.json b/examples/config.json new file mode 100644 index 0000000..fb473e5 --- /dev/null +++ b/examples/config.json @@ -0,0 +1,19 @@ +{ + "default_device": "phone", + "devices": { + "phone": { + "ssh_target": "root@phone", + "username": "defaultuser", + "architecture": "aarch64", + "release": "live", + "user_bus_runtime_dir": "/run/user/100000", + "user_bus_address": "unix:path=/run/user/100000/dbus/user_bus_socket" + } + }, + "paths": { + "git_root": "/home/you/git", + "ssh_config": "/home/you/.ssh/config", + "local_sdk": "/srv/mer/sdks/sfossdk/sdk-chroot", + "osc_api_alias": "your-obs-alias" + } +} diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..0a4851f --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,20 @@ +[build-system] +requires = ["setuptools>=61"] +build-backend = "setuptools.build_meta" + +[project] +name = "sailfish-devel-mcp" +version = "0.1.0" +description = "Host-side MCP server for Sailfish OS development workflows" +readme = "README.md" +requires-python = ">=3.9" +license = { text = "0BSD" } +authors = [{ name = "Andrew Branson" }] +dependencies = [] + +[project.scripts] +sailfish-devel-mcp = "sailfish_devel_mcp.server:main" + +[tool.setuptools.packages.find] +where = ["src"] + diff --git a/src/sailfish_devel_mcp/__init__.py b/src/sailfish_devel_mcp/__init__.py new file mode 100644 index 0000000..86c7cbb --- /dev/null +++ b/src/sailfish_devel_mcp/__init__.py @@ -0,0 +1,4 @@ +"""Sailfish OS development MCP server.""" + +__version__ = "0.1.0" + diff --git a/src/sailfish_devel_mcp/__main__.py b/src/sailfish_devel_mcp/__main__.py new file mode 100644 index 0000000..c5795c5 --- /dev/null +++ b/src/sailfish_devel_mcp/__main__.py @@ -0,0 +1,6 @@ +from .server import main + + +if __name__ == "__main__": + main() + diff --git a/src/sailfish_devel_mcp/config.py b/src/sailfish_devel_mcp/config.py new file mode 100644 index 0000000..6515918 --- /dev/null +++ b/src/sailfish_devel_mcp/config.py @@ -0,0 +1,178 @@ +from __future__ import annotations + +from dataclasses import dataclass, field +import json +import os +from pathlib import Path +from typing import Any, Mapping + + +DEFAULT_RUNTIME_DIR = "/run/user/100000" +DEFAULT_BUS_ADDRESS = "unix:path=/run/user/100000/dbus/user_bus_socket" +BUNDLED_BUILD_HELPER = ( + Path(__file__).resolve().parent / "vendor" / "build_sailfishos.py" +) + + +@dataclass(frozen=True) +class DeviceConfig: + name: str + ssh_target: str + username: str = "defaultuser" + architecture: str = "" + release: str = "" + user_bus_runtime_dir: str = DEFAULT_RUNTIME_DIR + user_bus_address: str = DEFAULT_BUS_ADDRESS + + def public_dict(self) -> dict[str, object]: + return { + "name": self.name, + "ssh_target": self.ssh_target, + "username": self.username, + "architecture": self.architecture, + "release": self.release, + "user_bus_runtime_dir": self.user_bus_runtime_dir, + "user_bus_address": self.user_bus_address, + } + + +@dataclass(frozen=True) +class PathConfig: + git_root: Path = Path.home() / "git" + ssh_config: Path = Path.home() / ".ssh" / "config" + build_sailfishos: Path = BUNDLED_BUILD_HELPER + local_sdk: Path | None = None + osc_api_alias: str = "" + + def public_dict(self) -> dict[str, str | None]: + return { + "git_root": str(self.git_root), + "ssh_config": str(self.ssh_config), + "build_sailfishos": str(self.build_sailfishos), + "local_sdk": str(self.local_sdk) if self.local_sdk else None, + "osc_api_alias": self.osc_api_alias, + } + + +@dataclass(frozen=True) +class Config: + path: Path | None + default_device: str + devices: Mapping[str, DeviceConfig] = field(default_factory=dict) + paths: PathConfig = field(default_factory=PathConfig) + + def device(self, name: str | None = None) -> DeviceConfig: + key = name or self.default_device + if key in self.devices: + return self.devices[key] + if name and "@" in name: + return DeviceConfig(name=name, ssh_target=name) + raise KeyError(f"unknown Sailfish device: {key}") + + def public_dict(self) -> dict[str, Any]: + return { + "path": str(self.path) if self.path else None, + "default_device": self.default_device, + "devices": { + name: device.public_dict() for name, device in self.devices.items() + }, + "paths": self.paths.public_dict(), + } + + +def default_config_path() -> Path: + if "SAILFISH_MCP_CONFIG" in os.environ: + return Path(os.environ["SAILFISH_MCP_CONFIG"]).expanduser() + return Path.home() / ".config" / "sailfish-devel-mcp" / "config.json" + + +def load_config(path: str | os.PathLike[str] | None = None) -> Config: + config_path = Path(path).expanduser() if path else default_config_path() + raw: dict[str, Any] = {} + + if config_path.exists(): + with config_path.open("r", encoding="utf-8") as handle: + loaded = json.load(handle) + if not isinstance(loaded, dict): + raise ValueError(f"{config_path} must contain a JSON object") + raw = loaded + + devices = _load_devices(raw.get("devices", {})) + if not devices: + default_name = os.environ.get("SAILFISH_MCP_DEFAULT_DEVICE", "device") + devices = { + default_name: DeviceConfig( + name=default_name, + ssh_target=os.environ.get("SAILFISH_MCP_SSH_TARGET", "root@device"), + ) + } + + default_device = str(raw.get("default_device") or next(iter(devices))) + paths = _load_paths(raw.get("paths", {})) + return Config( + path=config_path if config_path.exists() else None, + default_device=default_device, + devices=devices, + paths=paths, + ) + + +def _load_devices(raw_devices: Any) -> dict[str, DeviceConfig]: + if not isinstance(raw_devices, dict): + raise ValueError("devices must be a JSON object") + + devices: dict[str, DeviceConfig] = {} + for name, value in raw_devices.items(): + if isinstance(value, str): + value = {"ssh_target": value} + if not isinstance(value, dict): + raise ValueError(f"device {name!r} must be a string or object") + ssh_target = str(value.get("ssh_target") or name) + devices[str(name)] = DeviceConfig( + name=str(name), + ssh_target=ssh_target, + username=str(value.get("username") or "defaultuser"), + architecture=str(value.get("architecture") or ""), + release=str(value.get("release") or ""), + user_bus_runtime_dir=str( + value.get("user_bus_runtime_dir") or DEFAULT_RUNTIME_DIR + ), + user_bus_address=str(value.get("user_bus_address") or DEFAULT_BUS_ADDRESS), + ) + return devices + + +def _load_paths(raw_paths: Any) -> PathConfig: + if not isinstance(raw_paths, dict): + raise ValueError("paths must be a JSON object") + defaults = PathConfig() + raw_local_sdk = raw_paths.get("local_sdk") or os.environ.get("SAILFISH_MCP_LOCAL_SDK") + return PathConfig( + git_root=Path( + str( + raw_paths.get("git_root") + or os.environ.get("SAILFISH_MCP_GIT_ROOT") + or defaults.git_root + ) + ).expanduser(), + ssh_config=Path( + str( + raw_paths.get("ssh_config") + or os.environ.get("SAILFISH_MCP_SSH_CONFIG") + or defaults.ssh_config + ) + ).expanduser(), + build_sailfishos=Path( + str( + raw_paths.get("build_sailfishos") + or os.environ.get("SAILFISH_MCP_BUILD_HELPER") + or defaults.build_sailfishos + ) + ).expanduser(), + local_sdk=Path(str(raw_local_sdk)).expanduser() if raw_local_sdk else None, + osc_api_alias=str( + raw_paths.get("osc_api_alias") + or os.environ.get("SAILFISH_MCP_OSC_API_ALIAS") + or defaults.osc_api_alias + ), + ) diff --git a/src/sailfish_devel_mcp/runner.py b/src/sailfish_devel_mcp/runner.py new file mode 100644 index 0000000..b0b88bf --- /dev/null +++ b/src/sailfish_devel_mcp/runner.py @@ -0,0 +1,115 @@ +from __future__ import annotations + +from dataclasses import dataclass +from pathlib import Path +import shlex +import subprocess +from typing import Iterable, Sequence + +from .config import DeviceConfig + + +@dataclass(frozen=True) +class CommandResult: + argv: tuple[str, ...] + returncode: int + stdout: str + stderr: str + + @property + def ok(self) -> bool: + return self.returncode == 0 + + def public_dict(self, limit: int = 20000) -> dict[str, object]: + stdout, stdout_truncated = truncate(self.stdout, limit) + stderr, stderr_truncated = truncate(self.stderr, limit) + return { + "argv": list(self.argv), + "returncode": self.returncode, + "stdout": stdout, + "stderr": stderr, + "stdout_truncated": stdout_truncated, + "stderr_truncated": stderr_truncated, + } + + +def truncate(text: str, limit: int) -> tuple[str, bool]: + if len(text) <= limit: + return text, False + return text[:limit] + f"\n[truncated after {limit} characters]", True + + +def run( + argv: Sequence[str], + *, + cwd: str | Path | None = None, + timeout: int = 60, +) -> CommandResult: + try: + completed = subprocess.run( + list(argv), + cwd=str(cwd) if cwd is not None else None, + text=True, + capture_output=True, + timeout=timeout, + check=False, + ) + return CommandResult( + argv=tuple(str(arg) for arg in argv), + returncode=completed.returncode, + stdout=completed.stdout, + stderr=completed.stderr, + ) + except subprocess.TimeoutExpired as exc: + return CommandResult( + argv=tuple(str(arg) for arg in argv), + returncode=124, + stdout=exc.stdout or "", + stderr=(exc.stderr or "") + f"\ncommand timed out after {timeout}s", + ) + + +def ssh_argv(device: DeviceConfig, ssh_config: Path | None, remote: str) -> list[str]: + argv = ["ssh"] + if ssh_config: + argv += ["-F", str(ssh_config)] + argv += [device.ssh_target, remote] + return argv + + +def scp_to_argv( + device: DeviceConfig, + ssh_config: Path | None, + local_path: Path, + remote_path: str, +) -> list[str]: + argv = ["scp"] + if ssh_config: + argv += ["-F", str(ssh_config)] + argv += [str(local_path), f"{device.ssh_target}:{remote_path}"] + return argv + + +def scp_from_argv( + device: DeviceConfig, + ssh_config: Path | None, + remote_path: str, + local_path: Path, +) -> list[str]: + argv = ["scp"] + if ssh_config: + argv += ["-F", str(ssh_config)] + argv += [f"{device.ssh_target}:{remote_path}", str(local_path)] + return argv + + +def remote_command(argv: Iterable[str]) -> str: + return shlex.join([str(arg) for arg in argv]) + + +def user_bus_env(device: DeviceConfig) -> list[str]: + return [ + "env", + f"XDG_RUNTIME_DIR={device.user_bus_runtime_dir}", + f"DBUS_SESSION_BUS_ADDRESS={device.user_bus_address}", + ] diff --git a/src/sailfish_devel_mcp/server.py b/src/sailfish_devel_mcp/server.py new file mode 100644 index 0000000..4fb2f91 --- /dev/null +++ b/src/sailfish_devel_mcp/server.py @@ -0,0 +1,196 @@ +from __future__ import annotations + +import argparse +import json +import sys +from typing import Any, TextIO + +from . import __version__ +from .config import Config, load_config +from .tools import build_registry, tool_error + + +PROTOCOL_VERSIONS = [ + "2025-11-25", + "2025-06-18", + "2025-03-26", + "2024-11-05", +] + + +class JsonRpcError(Exception): + def __init__(self, code: int, message: str, data: Any | None = None): + super().__init__(message) + self.code = code + self.message = message + self.data = data + + +class McpServer: + def __init__(self, config: Config): + self.config = config + self.registry = build_registry(config) + + def handle(self, message: dict[str, Any]) -> dict[str, Any] | None: + if not isinstance(message, dict): + raise JsonRpcError(-32600, "JSON-RPC message must be an object") + + request_id = message.get("id") + method = message.get("method") + if not method: + raise JsonRpcError(-32600, "JSON-RPC message is missing method") + + if request_id is None: + self._handle_notification(method) + return None + + try: + result = self._dispatch(method, message.get("params") or {}) + return {"jsonrpc": "2.0", "id": request_id, "result": result} + except JsonRpcError as exc: + error: dict[str, Any] = {"code": exc.code, "message": exc.message} + if exc.data is not None: + error["data"] = exc.data + return {"jsonrpc": "2.0", "id": request_id, "error": error} + except Exception as exc: # pragma: no cover - defensive protocol boundary + return { + "jsonrpc": "2.0", + "id": request_id, + "error": {"code": -32603, "message": str(exc)}, + } + + def _handle_notification(self, method: str) -> None: + if method in {"notifications/initialized", "notifications/cancelled"}: + return + + def _dispatch(self, method: str, params: Any) -> dict[str, Any]: + if method == "initialize": + return self._initialize(params) + if method == "ping": + return {} + if method == "tools/list": + return {"tools": [tool.spec for tool in self.registry.values()]} + if method == "tools/call": + return self._call_tool(params) + if method == "resources/list": + return self._resources_list() + if method == "resources/read": + return self._resources_read(params) + if method == "prompts/list": + return {"prompts": []} + if method == "logging/setLevel": + return {} + raise JsonRpcError(-32601, f"method not found: {method}") + + def _initialize(self, params: Any) -> dict[str, Any]: + requested = "" + if isinstance(params, dict): + requested = str(params.get("protocolVersion") or "") + protocol = requested if requested in PROTOCOL_VERSIONS else PROTOCOL_VERSIONS[0] + return { + "protocolVersion": protocol, + "capabilities": { + "tools": {"listChanged": False}, + "resources": {"subscribe": False, "listChanged": False}, + "prompts": {"listChanged": False}, + }, + "serverInfo": { + "name": "sailfish-devel-mcp", + "version": __version__, + }, + "instructions": ( + "Host-side Sailfish OS development tools for devices, builds, " + "OBS, packaging, repositories, and QML checks." + ), + } + + def _call_tool(self, params: Any) -> dict[str, Any]: + if not isinstance(params, dict): + raise JsonRpcError(-32602, "tools/call params must be an object") + name = params.get("name") + if not isinstance(name, str): + raise JsonRpcError(-32602, "tools/call requires a tool name") + if name not in self.registry: + raise JsonRpcError(-32602, f"unknown tool: {name}") + args = params.get("arguments") or {} + if not isinstance(args, dict): + return tool_error("tool arguments must be an object") + try: + return self.registry[name].handler(args) + except ValueError as exc: + return tool_error(str(exc)) + + def _resources_list(self) -> dict[str, Any]: + return { + "resources": [ + { + "uri": "sailfish-devel-mcp://config/effective", + "name": "Effective configuration", + "mimeType": "application/json", + "description": "Resolved server configuration without secrets.", + }, + { + "uri": "sailfish-devel-mcp://help/tools", + "name": "Tool summary", + "mimeType": "text/plain", + "description": "Names and descriptions of exposed tools.", + }, + ] + } + + def _resources_read(self, params: Any) -> dict[str, Any]: + if not isinstance(params, dict) or not isinstance(params.get("uri"), str): + raise JsonRpcError(-32602, "resources/read requires a uri") + uri = params["uri"] + if uri == "sailfish-devel-mcp://config/effective": + text = json.dumps(self.config.public_dict(), indent=2, sort_keys=True) + return {"contents": [{"uri": uri, "mimeType": "application/json", "text": text}]} + if uri == "sailfish-devel-mcp://help/tools": + text = "\n".join( + f"{tool.spec['name']}: {tool.spec.get('description', '')}" + for tool in self.registry.values() + ) + return {"contents": [{"uri": uri, "mimeType": "text/plain", "text": text}]} + raise JsonRpcError(-32602, f"unknown resource: {uri}") + + +def run_stdio(server: McpServer, stdin: TextIO = sys.stdin, stdout: TextIO = sys.stdout) -> None: + for line in stdin: + if not line.strip(): + continue + try: + message = json.loads(line) + except json.JSONDecodeError as exc: + response = { + "jsonrpc": "2.0", + "id": None, + "error": {"code": -32700, "message": f"parse error: {exc}"}, + } + else: + response = server.handle(message) + if response is not None: + stdout.write(json.dumps(response, separators=(",", ":")) + "\n") + stdout.flush() + + +def main(argv: list[str] | None = None) -> None: + parser = argparse.ArgumentParser(description="Sailfish OS development MCP server") + parser.add_argument("--config", help="Path to config.json") + parser.add_argument( + "--dump-config", + action="store_true", + help="Print the resolved config and exit", + ) + args = parser.parse_args(argv) + + config = load_config(args.config) + if args.dump_config: + print(json.dumps(config.public_dict(), indent=2, sort_keys=True)) + return + + run_stdio(McpServer(config)) + + +if __name__ == "__main__": + main() + diff --git a/src/sailfish_devel_mcp/tools.py b/src/sailfish_devel_mcp/tools.py new file mode 100644 index 0000000..1061e6c --- /dev/null +++ b/src/sailfish_devel_mcp/tools.py @@ -0,0 +1,1282 @@ +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime +from pathlib import Path, PurePosixPath +import re +import shlex +import shutil +from typing import Any, Callable + +from .config import Config, DeviceConfig +from .runner import ( + CommandResult, + remote_command, + run, + scp_from_argv, + scp_to_argv, + ssh_argv, + truncate, + user_bus_env, +) + + +ToolHandler = Callable[[dict[str, Any]], dict[str, Any]] + + +@dataclass(frozen=True) +class Tool: + spec: dict[str, Any] + handler: ToolHandler + + +def build_registry(config: Config) -> dict[str, Tool]: + tools = [ + Tool(_spec_devices(), lambda args: handle_devices(config, args)), + Tool(_spec_device_journal(), lambda args: handle_device_journal(config, args)), + Tool(_spec_device_topmost_pid(), lambda args: handle_device_topmost_pid(config, args)), + Tool(_spec_device_proc_maps(), lambda args: handle_device_proc_maps(config, args)), + Tool( + _spec_device_lipstick_screenshot(), + lambda args: handle_device_lipstick_screenshot(config, args), + ), + Tool( + _spec_device_touch(), + lambda args: handle_device_touch(config, args), + ), + Tool( + _spec_device_user_bus_call(), + lambda args: handle_device_user_bus_call(config, args), + ), + Tool(_spec_device_install_rpm(), lambda args: handle_device_install_rpm(config, args)), + Tool( + _spec_device_restart_service(), + lambda args: handle_device_restart_service(config, args), + ), + Tool(_spec_build_rpm(), lambda args: handle_build_rpm(config, args)), + Tool(_spec_obs_results(), lambda args: handle_obs_results(config, args)), + Tool(_spec_obs_buildlog(), lambda args: handle_obs_buildlog(config, args)), + Tool(_spec_repo_status(), lambda args: handle_repo_status(config, args)), + Tool(_spec_repo_find(), lambda args: handle_repo_find(config, args)), + Tool(_spec_spec_summary(), lambda args: handle_spec_summary(config, args)), + Tool( + _spec_qml_find_translations(), + lambda args: handle_qml_find_translations(config, args), + ), + Tool( + _spec_qml_check_translator_ternaries(), + lambda args: handle_qml_check_translator_ternaries(config, args), + ), + ] + return {tool.spec["name"]: tool for tool in tools} + + +def handle_devices(config: Config, args: dict[str, Any]) -> dict[str, Any]: + structured = config.public_dict() + text = "\n".join( + ( + f"{name}: {device.ssh_target}" + f" user={device.username}" + f"{f' arch={device.architecture}' if device.architecture else ''}" + f"{f' release={device.release}' if device.release else ''}" + ) + for name, device in sorted(config.devices.items(), key=lambda item: item[0]) + ) + return ok_text(text or "No configured devices", structured) + + +def handle_device_journal(config: Config, args: dict[str, Any]) -> dict[str, Any]: + device = _device(config, args) + lines = _int_arg(args, "lines", default=200, minimum=1, maximum=5000) + timeout = _int_arg(args, "timeout", default=30, minimum=1, maximum=600) + + command = ["journalctl", "--no-pager", "-n", str(lines)] + unit = _optional_str(args, "unit") + since = _optional_str(args, "since") + grep = _optional_str(args, "grep") + if unit: + command += ["-u", unit] + if since: + command += ["--since", since] + result = _run_ssh(config, device, command, timeout=timeout) + stdout = result.stdout + if grep: + stdout = "\n".join(line for line in stdout.splitlines() if grep in line) + result = CommandResult(result.argv, result.returncode, stdout, result.stderr) + return command_result("device journal", result) + + +def handle_device_topmost_pid(config: Config, args: dict[str, Any]) -> dict[str, Any]: + device = _device(config, args) + timeout = _int_arg(args, "timeout", default=20, minimum=1, maximum=120) + command = user_bus_env(device) + [ + "dbus-send", + "--session", + "--print-reply", + "--dest=org.nemomobile.lipstick", + "/", + "org.freedesktop.DBus.Properties.Get", + "string:org.nemomobile.compositor", + "string:privateTopmostWindowProcessId", + ] + result = _run_ssh(config, device, command, timeout=timeout) + pid = _parse_dbus_integer(result.stdout) + structured = result.public_dict() + structured["pid"] = pid + text = f"topmost PID: {pid}" if pid is not None else _command_text("topmost PID", result) + return { + "content": [{"type": "text", "text": text}], + "structuredContent": structured, + "isError": not result.ok or pid is None, + } + + +def handle_device_proc_maps(config: Config, args: dict[str, Any]) -> dict[str, Any]: + device = _device(config, args) + pid = _int_arg(args, "pid", minimum=1) + contains = _optional_str(args, "contains") + max_lines = _int_arg(args, "max_lines", default=200, minimum=1, maximum=5000) + timeout = _int_arg(args, "timeout", default=20, minimum=1, maximum=120) + + result = _run_ssh(config, device, ["cat", f"/proc/{pid}/maps"], timeout=timeout) + lines = result.stdout.splitlines() + if contains: + lines = [line for line in lines if contains in line] + limited = lines[:max_lines] + stdout = "\n".join(limited) + if len(lines) > max_lines: + stdout += f"\n[truncated after {max_lines} lines]" + filtered = CommandResult(result.argv, result.returncode, stdout, result.stderr) + return command_result("process maps", filtered, {"matched_lines": len(lines), "pid": pid}) + + +def handle_device_lipstick_screenshot(config: Config, args: dict[str, Any]) -> dict[str, Any]: + device = _device(config, args) + timestamp = datetime.utcnow().strftime("%Y%m%d-%H%M%S") + home_path = _device_home_path(device) + remote_path = _optional_str(args, "remote_path") or ( + f"{home_path}/Pictures/Screenshots/lipstick-{timestamp}.png" + ) + local_path_arg = _optional_str(args, "local_path") + privileged = _bool_arg(args, "privileged", default=True) + timeout = _int_arg(args, "timeout", default=30, minimum=1, maximum=120) + + inner = remote_command( + user_bus_env(device) + + [ + "dbus-send", + "--session", + "--print-reply", + "--dest=org.nemomobile.lipstick", + "/org/nemomobile/lipstick/screenshot", + "org.nemomobile.lipstick.saveScreenshot", + f"string:{remote_path}", + ] + ) + prepare_dir = _screenshot_prepare_command(device, remote_path) + dbus_call = f"sg privileged -c {shlex.quote(inner)}" if privileged else inner + remote = f"{prepare_dir} && {dbus_call}" + result = run(ssh_argv(device, config.paths.ssh_config, remote), timeout=timeout) + + structured: dict[str, Any] = result.public_dict() + structured["remote_path"] = remote_path + if result.ok and local_path_arg: + local_path = _safe_output_path(config, local_path_arg) + local_path.parent.mkdir(parents=True, exist_ok=True) + pull = run( + scp_from_argv(device, config.paths.ssh_config, remote_path, local_path), + timeout=timeout, + ) + structured["pull"] = pull.public_dict() + structured["local_path"] = str(local_path) + if not pull.ok: + return command_result("pull screenshot", pull, structured) + return command_result("lipstick screenshot", result, structured) + + +def handle_device_touch(config: Config, args: dict[str, Any]) -> dict[str, Any]: + device = _device(config, args) + action = _enum_arg(args, "action", ["discover", "tap", "swipe"]) + timeout = _int_arg(args, "timeout", default=10, minimum=1, maximum=120) + input_device = _optional_str(args, "input_device") + if input_device and not re.fullmatch(r"/dev/input/event\d+", input_device): + return tool_error("input_device must look like /dev/input/event<N>") + + if action == "discover": + include_evdev_trace = _bool_arg(args, "include_evdev_trace", default=False) + remote = _touch_discover_command(include_evdev_trace) + return command_result( + "touchscreen discovery", + run(ssh_argv(device, config.paths.ssh_config, remote), timeout=timeout), + ) + + values: dict[str, str | int] = {"ACTION": action} + if input_device: + values["INPUT_DEVICE"] = input_device + + if action == "tap": + values["X"] = _int_arg(args, "x", minimum=0, maximum=10000) + values["Y"] = _int_arg(args, "y", minimum=0, maximum=10000) + values["HOLD_MS"] = _int_arg(args, "hold_ms", default=80, minimum=1, maximum=5000) + else: + values["START_X"] = _int_arg(args, "start_x", minimum=0, maximum=10000) + values["START_Y"] = _int_arg(args, "start_y", minimum=0, maximum=10000) + values["END_X"] = _int_arg(args, "end_x", minimum=0, maximum=10000) + values["END_Y"] = _int_arg(args, "end_y", minimum=0, maximum=10000) + values["DURATION_MS"] = _int_arg(args, "duration_ms", default=300, minimum=1, maximum=10000) + values["STEPS"] = _int_arg(args, "steps", default=12, minimum=1, maximum=200) + + remote = _touch_inject_command(values) + structured = { + "action": action, + "input_device": input_device, + } + return command_result( + "device touch", + run(ssh_argv(device, config.paths.ssh_config, remote), timeout=timeout), + structured, + ) + + +def handle_device_user_bus_call(config: Config, args: dict[str, Any]) -> dict[str, Any]: + device = _device(config, args) + destination = _str_arg(args, "destination") + path = _str_arg(args, "path") + interface = _str_arg(args, "interface") + member = _str_arg(args, "member") + dbus_args = args.get("arguments") or [] + if not isinstance(dbus_args, list) or not all(isinstance(item, str) for item in dbus_args): + return tool_error("arguments must be a list of dbus-send argument strings") + timeout = _int_arg(args, "timeout", default=30, minimum=1, maximum=300) + command = user_bus_env(device) + [ + "dbus-send", + "--session", + "--print-reply", + f"--dest={destination}", + path, + f"{interface}.{member}", + *dbus_args, + ] + return command_result("user bus call", _run_ssh(config, device, command, timeout=timeout)) + + +def handle_device_install_rpm(config: Config, args: dict[str, Any]) -> dict[str, Any]: + device = _device(config, args) + rpm_path = _safe_input_path(config, _str_arg(args, "rpm_path"), allow_tmp=True) + remote_path = _optional_str(args, "remote_path") or f"/tmp/{rpm_path.name}" + installer = _enum_arg(args, "installer", ["pkcon", "rpm"], default="pkcon") + timeout = _int_arg(args, "timeout", default=180, minimum=1, maximum=1200) + + copy_result = run( + scp_to_argv(device, config.paths.ssh_config, rpm_path, remote_path), + timeout=timeout, + ) + if not copy_result.ok: + return command_result("copy RPM to device", copy_result) + + if installer == "pkcon": + remote = ["pkcon", "install-local", "-y", remote_path] + else: + remote = ["rpm", "-Uvh", "--replacepkgs", remote_path] + install_result = _run_ssh(config, device, remote, timeout=timeout) + structured = { + "copy": copy_result.public_dict(), + "install": install_result.public_dict(), + "remote_path": remote_path, + "installer": installer, + } + return command_result("install RPM", install_result, structured) + + +def handle_device_restart_service(config: Config, args: dict[str, Any]) -> dict[str, Any]: + device = _device(config, args) + unit = _service_unit_arg(args, "unit") + action = _enum_arg(args, "action", ["restart", "start", "stop", "status"], default="restart") + mode = _enum_arg(args, "mode", ["system", "user"], default="system") + timeout = _int_arg(args, "timeout", default=60, minimum=1, maximum=300) + if mode == "user": + command = user_bus_env(device) + ["systemctl", "--user", action, unit] + else: + command = ["systemctl", action, unit] + return command_result( + f"{mode} service {action}", + _run_ssh(config, device, command, timeout=timeout), + ) + + +def handle_build_rpm(config: Config, args: dict[str, Any]) -> dict[str, Any]: + project_path = _safe_input_path(config, _str_arg(args, "project_path"), allow_tmp=False) + device = config.device(_optional_str(args, "device")) if args.get("device") else None + script = config.paths.build_sailfishos + if not script.exists(): + return tool_error(f"build helper not found: {script}") + + command = ["python3", str(script), "--project-dir", str(project_path)] + release = _optional_str(args, "release") or (device.release if device else None) + arches = args.get("arch") + artifacts_dir = _optional_str(args, "artifacts_dir") + if config.paths.local_sdk: + command += ["--local-sdk", str(config.paths.local_sdk)] + if release: + command += ["--release", release] + if isinstance(arches, str): + command += ["--arch", arches] + elif isinstance(arches, list): + for arch in arches: + if not isinstance(arch, str): + return tool_error("arch must be a string or list of strings") + command += ["--arch", arch] + elif arches is not None: + return tool_error("arch must be a string or list of strings") + elif device and device.architecture: + command += ["--arch", device.architecture] + if artifacts_dir: + output = _safe_output_path(config, artifacts_dir) + command += ["--artifacts-dir", str(output)] + if _bool_arg(args, "all_arches", default=False): + command.append("--all") + if _bool_arg(args, "clean", default=False): + command.append("--clean") + if _bool_arg(args, "debug", default=False): + command.append("--debug") + if _bool_arg(args, "no_pull", default=False): + command.append("--no-pull") + for local_dir in _string_list_arg(args, "local_rpms_dir"): + command += ["--local-rpms-dir", str(_safe_input_path(config, local_dir, allow_tmp=True))] + timeout = _int_arg(args, "timeout", default=3600, minimum=1, maximum=21600) + return command_result("build Sailfish RPM", run(command, timeout=timeout)) + + +def handle_obs_results(config: Config, args: dict[str, Any]) -> dict[str, Any]: + project = _str_arg(args, "project") + package = _optional_str(args, "package") + api_alias = _optional_str(args, "api_alias") or config.paths.osc_api_alias + command = ["osc"] + if api_alias: + command += ["-A", api_alias] + command += ["results", project] + if package: + command.append(package) + timeout = _int_arg(args, "timeout", default=60, minimum=1, maximum=600) + return command_result("OBS results", run(command, timeout=timeout)) + + +def handle_obs_buildlog(config: Config, args: dict[str, Any]) -> dict[str, Any]: + project = _str_arg(args, "project") + package = _str_arg(args, "package") + repository = _str_arg(args, "repository") + arch = _str_arg(args, "arch") + api_alias = _optional_str(args, "api_alias") or config.paths.osc_api_alias + timeout = _int_arg(args, "timeout", default=90, minimum=1, maximum=1800) + command = ["osc"] + if api_alias: + command += ["-A", api_alias] + command += ["remotebuildlog", project, package, repository, arch] + return command_result("OBS build log", run(command, timeout=timeout)) + + +def handle_repo_status(config: Config, args: dict[str, Any]) -> dict[str, Any]: + path = _safe_input_path(config, _optional_str(args, "path") or ".", allow_tmp=False) + timeout = _int_arg(args, "timeout", default=30, minimum=1, maximum=120) + result = run(["git", "status", "--short", "--branch"], cwd=path, timeout=timeout) + return command_result("git status", result, {"path": str(path)}) + + +def handle_repo_find(config: Config, args: dict[str, Any]) -> dict[str, Any]: + path = _safe_input_path(config, _optional_str(args, "path") or ".", allow_tmp=False) + query = _str_arg(args, "query") + max_count = _int_arg(args, "max_count", default=100, minimum=1, maximum=1000) + fixed_strings = _bool_arg(args, "fixed_strings", default=True) + timeout = _int_arg(args, "timeout", default=30, minimum=1, maximum=300) + + if shutil.which("rg"): + command = ["rg", "--line-number", "--hidden", "--glob", "!.git"] + if fixed_strings: + command.append("--fixed-strings") + command += ["--max-count", str(max_count), query, str(path)] + else: + command = ["grep", "-R", "-n", query, str(path)] + result = run(command, timeout=timeout) + if result.returncode == 1 and not result.stdout: + return ok_text("no matches", {"path": str(path), "query": query, "matches": 0}) + return command_result("repo find", result, {"path": str(path), "query": query}) + + +def handle_spec_summary(config: Config, args: dict[str, Any]) -> dict[str, Any]: + path_arg = _optional_str(args, "spec_path") + if path_arg: + spec_path = _safe_input_path(config, path_arg, allow_tmp=False) + else: + repo = _safe_input_path(config, _optional_str(args, "repo_path") or ".", allow_tmp=False) + specs = sorted((repo / "rpm").glob("*.spec")) + if not specs: + return tool_error(f"no rpm/*.spec file found under {repo}") + spec_path = specs[0] + + if not spec_path.exists(): + return tool_error(f"spec file does not exist: {spec_path}") + data = _parse_spec(spec_path) + text_lines = [f"{key}: {value}" for key, value in data.items() if value] + return ok_text("\n".join(text_lines), {"spec_path": str(spec_path), "summary": data}) + + +def handle_qml_find_translations(config: Config, args: dict[str, Any]) -> dict[str, Any]: + path = _safe_input_path(config, _optional_str(args, "path") or ".", allow_tmp=False) + timeout = _int_arg(args, "timeout", default=30, minimum=1, maximum=300) + if not shutil.which("rg"): + return tool_error("rg is required for qml_find_translations") + command = [ + "rg", + "--line-number", + "--glob", + "*.qml", + r"qsTrId|//%|//:", + str(path), + ] + result = run(command, timeout=timeout) + if result.returncode == 1 and not result.stdout: + return ok_text("no QML translation markers found", {"path": str(path), "matches": 0}) + return command_result("QML translations", result, {"path": str(path)}) + + +def handle_qml_check_translator_ternaries(config: Config, args: dict[str, Any]) -> dict[str, Any]: + path = _safe_input_path(config, _optional_str(args, "path") or ".", allow_tmp=False) + files = [path] if path.is_file() and path.suffix == ".qml" else sorted(path.rglob("*.qml")) + findings: list[dict[str, Any]] = [] + for qml in files: + if ".git" in qml.parts: + continue + try: + lines = qml.read_text(encoding="utf-8").splitlines() + except UnicodeDecodeError: + continue + for index, line in enumerate(lines, start=1): + if "qsTrId(" in line and "?" in line and ":" in line: + findings.append( + { + "path": str(qml), + "line": index, + "text": line.strip(), + "message": ( + "ternary qsTrId expression should give each branch " + "its own translator comment and source text" + ), + } + ) + if not findings: + return ok_text("no ternary translation issues found", {"path": str(path), "findings": []}) + text = "\n".join( + f"{item['path']}:{item['line']}: {item['message']}\n {item['text']}" + for item in findings + ) + return { + "content": [{"type": "text", "text": text}], + "structuredContent": {"path": str(path), "findings": findings}, + "isError": True, + } + + +def _run_ssh( + config: Config, + device: DeviceConfig, + command: list[str], + *, + timeout: int, +) -> CommandResult: + return run( + ssh_argv(device, config.paths.ssh_config, remote_command(command)), + timeout=timeout, + ) + + +def _screenshot_prepare_command(device: DeviceConfig, remote_path: str) -> str: + remote_dir = str(PurePosixPath(remote_path).parent) + home_path = _device_home_path(device) + script = f""" +set -e +home={shlex.quote(home_path)} +remote_dir={shlex.quote(remote_dir)} +owner=$(stat -Lc %U "$home") +group=$(stat -Lc %G "$home") +if [ "$remote_dir" = "$home/Pictures/Screenshots" ]; then + if [ ! -d "$home/Pictures" ]; then + install -d -m 775 -o "$owner" -g "$group" "$home/Pictures" + fi + if getent group privileged >/dev/null 2>&1; then + screenshot_group=privileged + else + screenshot_group=$group + fi + if [ ! -d "$remote_dir" ]; then + install -d -m 755 -o "$owner" -g "$screenshot_group" "$remote_dir" + fi +else + if [ ! -d "$remote_dir" ]; then + install -d -m 755 -o "$owner" -g "$group" "$remote_dir" + fi +fi +""".strip() + return remote_command(["sh", "-lc", script]) + + +def _touch_discover_command(include_evdev_trace: bool) -> str: + evdev_trace = """ +if command -v evdev_trace >/dev/null 2>&1; then + echo "### evdev_trace -i" + if command -v timeout >/dev/null 2>&1; then + timeout 5 evdev_trace -i || true + else + evdev_trace -i & + pid=$! + sleep 5 + kill "$pid" >/dev/null 2>&1 || true + wait "$pid" >/dev/null 2>&1 || true + fi + echo +fi +""".strip() + script = f""" +set -e +{evdev_trace if include_evdev_trace else ""} +echo "### /proc/bus/input/devices" +cat /proc/bus/input/devices +""".strip() + return remote_command(["sh", "-lc", script]) + + +def _touch_inject_command(values: dict[str, str | int]) -> str: + env = " ".join( + f"{key}={shlex.quote(str(value))}" + for key, value in sorted(values.items()) + ) + script = f""" +set -e +{env} python3 - <<'PY' +{_touch_inject_python()} +PY +""".strip() + return remote_command(["sh", "-lc", script]) + + +def _touch_inject_python() -> str: + return r''' +import os +import re +import struct +import sys +import time + +EV_SYN = 0 +EV_KEY = 1 +EV_ABS = 3 +SYN_REPORT = 0 +BTN_TOUCH = 0x14a +ABS_X = 0x00 +ABS_Y = 0x01 +ABS_MT_SLOT = 0x2f +ABS_MT_POSITION_X = 0x35 +ABS_MT_POSITION_Y = 0x36 +ABS_MT_TRACKING_ID = 0x39 + +TOUCH_KEYWORDS = ( + "touch", + "touchscreen", + "digitizer", + "fts", + "ft5x", + "goodix", + "synaptics", + "cyttsp", + "atmel", + "elan", + "himax", + "novatek", + "nvt", + "silead", + "gt9", + "spi3", +) + + +def die(message): + print(message, file=sys.stderr) + raise SystemExit(2) + + +def input_blocks(): + try: + text = open("/proc/bus/input/devices", encoding="utf-8").read() + except OSError as exc: + die(f"failed to read /proc/bus/input/devices: {exc}") + return [block for block in text.split("\n\n") if block.strip()] + + +def field(block, prefix): + for line in block.splitlines(): + if line.startswith(prefix): + return line[len(prefix):].strip() + return "" + + +def event_handler(block): + match = re.search(r"\bevent\d+\b", field(block, "H: Handlers=")) + return match.group(0) if match else "" + + +def name(block): + value = field(block, "N: Name=") + return value.strip('"') + + +def score(block): + lower = block.lower() + value = 0 + if any(keyword in lower for keyword in TOUCH_KEYWORDS): + value += 100 + if "b: abs=" in lower: + value += 20 + if "b: prop=" in lower: + value += 5 + if "mouse" in lower or "keyboard" in lower or "keypad" in lower: + value -= 100 + return value + + +def discover(): + devices = [] + for block in input_blocks(): + event = event_handler(block) + if not event: + continue + devices.append( + { + "path": f"/dev/input/{event}", + "name": name(block), + "score": score(block), + } + ) + devices.sort(key=lambda item: item["score"], reverse=True) + return devices + + +def choose_device(): + explicit = os.environ.get("INPUT_DEVICE", "") + if explicit: + return explicit + devices = discover() + if not devices or devices[0]["score"] <= 0: + summary = ", ".join(f"{item['path']}:{item['name']}" for item in devices) + die(f"could not identify touchscreen input device; candidates: {summary}") + return devices[0]["path"] + + +def event(fileobj, event_type, code, value): + fileobj.write(struct.pack("@llHHi", 0, 0, event_type, code, value)) + + +def sync(fileobj): + event(fileobj, EV_SYN, SYN_REPORT, 0) + + +def move(fileobj, x, y): + event(fileobj, EV_ABS, ABS_MT_POSITION_X, x) + event(fileobj, EV_ABS, ABS_MT_POSITION_Y, y) + event(fileobj, EV_ABS, ABS_X, x) + event(fileobj, EV_ABS, ABS_Y, y) + + +def down(fileobj, x, y): + event(fileobj, EV_ABS, ABS_MT_SLOT, 0) + event(fileobj, EV_ABS, ABS_MT_TRACKING_ID, int(time.time() * 1000) & 0x7fffffff) + move(fileobj, x, y) + event(fileobj, EV_KEY, BTN_TOUCH, 1) + sync(fileobj) + + +def up(fileobj): + event(fileobj, EV_ABS, ABS_MT_SLOT, 0) + event(fileobj, EV_ABS, ABS_MT_TRACKING_ID, -1) + event(fileobj, EV_KEY, BTN_TOUCH, 0) + sync(fileobj) + + +def integer(name): + try: + return int(os.environ[name]) + except KeyError: + die(f"missing {name}") + except ValueError: + die(f"{name} must be an integer") + + +action = os.environ.get("ACTION", "") +device = choose_device() +print(f"input_device={device}") +with open(device, "wb", buffering=0) as fileobj: + if action == "tap": + down(fileobj, integer("X"), integer("Y")) + time.sleep(integer("HOLD_MS") / 1000) + up(fileobj) + elif action == "swipe": + start_x = integer("START_X") + start_y = integer("START_Y") + end_x = integer("END_X") + end_y = integer("END_Y") + steps = integer("STEPS") + duration = integer("DURATION_MS") / 1000 + down(fileobj, start_x, start_y) + for step in range(1, steps + 1): + fraction = step / steps + x = round(start_x + ((end_x - start_x) * fraction)) + y = round(start_y + ((end_y - start_y) * fraction)) + move(fileobj, x, y) + sync(fileobj) + time.sleep(duration / steps) + up(fileobj) + else: + die(f"unsupported ACTION {action!r}") +print(f"{action}=ok") +'''.strip() + + +def _device_home_path(device: DeviceConfig) -> str: + return f"/home/{device.username}" + + +def command_result( + title: str, + result: CommandResult, + structured: dict[str, Any] | None = None, +) -> dict[str, Any]: + data = result.public_dict() + if structured: + data.update(structured) + return { + "content": [{"type": "text", "text": _command_text(title, result)}], + "structuredContent": data, + "isError": not result.ok, + } + + +def ok_text(text: str, structured: dict[str, Any] | None = None) -> dict[str, Any]: + return { + "content": [{"type": "text", "text": text}], + "structuredContent": structured or {}, + "isError": False, + } + + +def tool_error(message: str, structured: dict[str, Any] | None = None) -> dict[str, Any]: + return { + "content": [{"type": "text", "text": message}], + "structuredContent": structured or {"error": message}, + "isError": True, + } + + +def _command_text(title: str, result: CommandResult) -> str: + stdout, _ = truncate(result.stdout, 12000) + stderr, _ = truncate(result.stderr, 8000) + parts = [f"{title}: exit {result.returncode}"] + if stdout: + parts += ["", stdout.rstrip()] + if stderr: + parts += ["", "stderr:", stderr.rstrip()] + return "\n".join(parts) + + +def _device(config: Config, args: dict[str, Any]) -> DeviceConfig: + return config.device(_optional_str(args, "device")) + + +def _str_arg(args: dict[str, Any], name: str) -> str: + value = args.get(name) + if not isinstance(value, str) or not value: + raise ValueError(f"{name} must be a non-empty string") + return value + + +def _optional_str(args: dict[str, Any], name: str) -> str | None: + value = args.get(name) + if value is None or value == "": + return None + if not isinstance(value, str): + raise ValueError(f"{name} must be a string") + return value + + +def _int_arg( + args: dict[str, Any], + name: str, + default: int | None = None, + minimum: int | None = None, + maximum: int | None = None, +) -> int: + value = args.get(name, default) + if not isinstance(value, int): + raise ValueError(f"{name} must be an integer") + if minimum is not None and value < minimum: + raise ValueError(f"{name} must be >= {minimum}") + if maximum is not None and value > maximum: + raise ValueError(f"{name} must be <= {maximum}") + return value + + +def _bool_arg(args: dict[str, Any], name: str, default: bool = False) -> bool: + value = args.get(name, default) + if not isinstance(value, bool): + raise ValueError(f"{name} must be a boolean") + return value + + +def _enum_arg( + args: dict[str, Any], + name: str, + values: list[str], + default: str | None = None, +) -> str: + value = args.get(name, default) + if not isinstance(value, str) or value not in values: + raise ValueError(f"{name} must be one of: {', '.join(values)}") + return value + + +def _string_list_arg(args: dict[str, Any], name: str) -> list[str]: + value = args.get(name) + if value is None: + return [] + if not isinstance(value, list) or not all(isinstance(item, str) for item in value): + raise ValueError(f"{name} must be a list of strings") + return value + + +def _service_unit_arg(args: dict[str, Any], name: str) -> str: + unit = _str_arg(args, name) + if not re.fullmatch(r"[A-Za-z0-9@_.:\-]+", unit): + raise ValueError(f"{name} contains unsupported characters") + return unit + + +def _safe_input_path(config: Config, value: str, *, allow_tmp: bool) -> Path: + path = Path(value).expanduser() + if not path.is_absolute(): + path = config.paths.git_root / path + path = path.resolve(strict=False) + roots = [config.paths.git_root.resolve(strict=False)] + if allow_tmp: + roots.append(Path("/tmp").resolve(strict=False)) + if not _is_relative_to_any(path, roots): + raise ValueError(f"path is outside allowed roots: {path}") + return path + + +def _safe_output_path(config: Config, value: str) -> Path: + path = Path(value).expanduser() + if not path.is_absolute(): + path = config.paths.git_root / path + path = path.resolve(strict=False) + roots = [config.paths.git_root.resolve(strict=False), Path("/tmp").resolve(strict=False)] + if not _is_relative_to_any(path, roots): + raise ValueError(f"output path is outside allowed roots: {path}") + return path + + +def _is_relative_to_any(path: Path, roots: list[Path]) -> bool: + for root in roots: + try: + path.relative_to(root) + return True + except ValueError: + pass + return False + + +def _parse_dbus_integer(text: str) -> int | None: + match = re.search(r"\b(?:int32|uint32|int64|uint64)\s+(-?\d+)", text) + return int(match.group(1)) if match else None + + +def _parse_spec(path: Path) -> dict[str, Any]: + fields: dict[str, Any] = { + "Name": "", + "Version": "", + "Release": "", + "Summary": "", + "License": "", + "URL": "", + "BuildRequires": [], + "Requires": [], + } + for raw_line in path.read_text(encoding="utf-8", errors="replace").splitlines(): + line = raw_line.strip() + if not line or line.startswith("#"): + continue + for field in ["Name", "Version", "Release", "Summary", "License", "URL"]: + prefix = field + ":" + if line.startswith(prefix) and not fields[field]: + fields[field] = line[len(prefix) :].strip() + if line.startswith("BuildRequires:"): + fields["BuildRequires"].append(line.split(":", 1)[1].strip()) + if line.startswith("Requires:"): + fields["Requires"].append(line.split(":", 1)[1].strip()) + return fields + + +def _read_only_annotations(title: str) -> dict[str, Any]: + return { + "title": title, + "readOnlyHint": True, + "destructiveHint": False, + "idempotentHint": True, + "openWorldHint": True, + } + + +def _mutating_annotations(title: str, *, destructive: bool = False) -> dict[str, Any]: + return { + "title": title, + "readOnlyHint": False, + "destructiveHint": destructive, + "idempotentHint": False, + "openWorldHint": True, + } + + +def _object_schema(properties: dict[str, Any], required: list[str] | None = None) -> dict[str, Any]: + return { + "type": "object", + "properties": properties, + "required": required or [], + "additionalProperties": False, + } + + +def _device_prop() -> dict[str, Any]: + return {"type": "string", "description": "Configured device alias or ssh target."} + + +def _timeout_prop(default: int) -> dict[str, Any]: + return { + "type": "integer", + "minimum": 1, + "default": default, + "description": "Command timeout in seconds.", + } + + +def _spec_devices() -> dict[str, Any]: + return { + "name": "sailfish_devices", + "title": "List Sailfish Devices", + "description": "List configured Sailfish OS device aliases and SSH targets.", + "inputSchema": _object_schema({}), + "annotations": _read_only_annotations("List Sailfish Devices"), + } + + +def _spec_device_journal() -> dict[str, Any]: + return { + "name": "sailfish_device_journal", + "title": "Read Device Journal", + "description": "Read recent journalctl output from a Sailfish OS device.", + "inputSchema": _object_schema( + { + "device": _device_prop(), + "unit": {"type": "string"}, + "since": {"type": "string"}, + "grep": {"type": "string"}, + "lines": {"type": "integer", "minimum": 1, "maximum": 5000, "default": 200}, + "timeout": _timeout_prop(30), + } + ), + "annotations": _read_only_annotations("Read Device Journal"), + } + + +def _spec_device_topmost_pid() -> dict[str, Any]: + return { + "name": "sailfish_device_topmost_pid", + "title": "Topmost Window PID", + "description": "Query Lipstick for the current topmost window process id.", + "inputSchema": _object_schema({"device": _device_prop(), "timeout": _timeout_prop(20)}), + "annotations": _read_only_annotations("Topmost Window PID"), + } + + +def _spec_device_proc_maps() -> dict[str, Any]: + return { + "name": "sailfish_device_proc_maps", + "title": "Read Process Maps", + "description": "Read or filter /proc/<pid>/maps on a Sailfish OS device.", + "inputSchema": _object_schema( + { + "device": _device_prop(), + "pid": {"type": "integer", "minimum": 1}, + "contains": {"type": "string"}, + "max_lines": {"type": "integer", "minimum": 1, "maximum": 5000, "default": 200}, + "timeout": _timeout_prop(20), + }, + ["pid"], + ), + "annotations": _read_only_annotations("Read Process Maps"), + } + + +def _spec_device_lipstick_screenshot() -> dict[str, Any]: + return { + "name": "sailfish_device_lipstick_screenshot", + "title": "Lipstick Screenshot", + "description": "Ask Lipstick to save a screenshot on the device, optionally pulling it locally.", + "inputSchema": _object_schema( + { + "device": _device_prop(), + "remote_path": { + "type": "string", + "default": "/home/defaultuser/Pictures/Screenshots/lipstick-<timestamp>.png", + "description": "Lipstick accepts screenshot paths under the user home directory.", + }, + "local_path": {"type": "string"}, + "privileged": {"type": "boolean", "default": True}, + "timeout": _timeout_prop(30), + } + ), + "annotations": _mutating_annotations("Lipstick Screenshot"), + } + + +def _spec_device_touch() -> dict[str, Any]: + return { + "name": "sailfish_device_touch", + "title": "Device Touch Input", + "description": "Discover the touchscreen input device or inject tap/swipe events over SSH.", + "inputSchema": _object_schema( + { + "device": _device_prop(), + "action": { + "type": "string", + "enum": ["discover", "tap", "swipe"], + "description": "discover lists input devices; tap and swipe inject Linux input events.", + }, + "include_evdev_trace": { + "type": "boolean", + "default": False, + "description": "Also run evdev_trace -i during discovery. Disabled by default because it may block on some devices.", + }, + "input_device": { + "type": "string", + "description": "Optional explicit device path, for example /dev/input/event5.", + }, + "x": {"type": "integer", "minimum": 0, "maximum": 10000}, + "y": {"type": "integer", "minimum": 0, "maximum": 10000}, + "hold_ms": {"type": "integer", "minimum": 1, "maximum": 5000, "default": 80}, + "start_x": {"type": "integer", "minimum": 0, "maximum": 10000}, + "start_y": {"type": "integer", "minimum": 0, "maximum": 10000}, + "end_x": {"type": "integer", "minimum": 0, "maximum": 10000}, + "end_y": {"type": "integer", "minimum": 0, "maximum": 10000}, + "duration_ms": {"type": "integer", "minimum": 1, "maximum": 10000, "default": 300}, + "steps": {"type": "integer", "minimum": 1, "maximum": 200, "default": 12}, + "timeout": _timeout_prop(10), + }, + ["action"], + ), + "annotations": _mutating_annotations("Device Touch Input"), + } + + +def _spec_device_user_bus_call() -> dict[str, Any]: + return { + "name": "sailfish_device_user_bus_call", + "title": "User Bus Call", + "description": "Run a typed dbus-send method call on the defaultuser session bus.", + "inputSchema": _object_schema( + { + "device": _device_prop(), + "destination": {"type": "string"}, + "path": {"type": "string"}, + "interface": {"type": "string"}, + "member": {"type": "string"}, + "arguments": { + "type": "array", + "items": {"type": "string"}, + "description": "Raw dbus-send argument strings, for example string:/tmp/file.", + }, + "timeout": _timeout_prop(30), + }, + ["destination", "path", "interface", "member"], + ), + "annotations": _mutating_annotations("User Bus Call"), + } + + +def _spec_device_install_rpm() -> dict[str, Any]: + return { + "name": "sailfish_device_install_rpm", + "title": "Install Device RPM", + "description": "Copy a local RPM to a Sailfish OS device and install it.", + "inputSchema": _object_schema( + { + "device": _device_prop(), + "rpm_path": {"type": "string"}, + "remote_path": {"type": "string"}, + "installer": {"type": "string", "enum": ["pkcon", "rpm"], "default": "pkcon"}, + "timeout": _timeout_prop(180), + }, + ["rpm_path"], + ), + "annotations": _mutating_annotations("Install Device RPM"), + } + + +def _spec_device_restart_service() -> dict[str, Any]: + return { + "name": "sailfish_device_restart_service", + "title": "Manage Device Service", + "description": "Run systemctl start/stop/restart/status for a system or user service.", + "inputSchema": _object_schema( + { + "device": _device_prop(), + "unit": {"type": "string"}, + "action": { + "type": "string", + "enum": ["restart", "start", "stop", "status"], + "default": "restart", + }, + "mode": {"type": "string", "enum": ["system", "user"], "default": "system"}, + "timeout": _timeout_prop(60), + }, + ["unit"], + ), + "annotations": _mutating_annotations("Manage Device Service"), + } + + +def _spec_build_rpm() -> dict[str, Any]: + return { + "name": "sailfish_build_rpm", + "title": "Build Sailfish RPM", + "description": "Run the local build-sailfishos helper; paths.local_sdk is used only when it has a matching target.", + "inputSchema": _object_schema( + { + "project_path": {"type": "string"}, + "device": { + "type": "string", + "description": "Optional configured device to supply default release and architecture.", + }, + "release": {"type": "string"}, + "arch": { + "oneOf": [ + {"type": "string"}, + {"type": "array", "items": {"type": "string"}}, + ] + }, + "artifacts_dir": {"type": "string"}, + "all_arches": {"type": "boolean", "default": False}, + "clean": {"type": "boolean", "default": False}, + "debug": {"type": "boolean", "default": False}, + "no_pull": {"type": "boolean", "default": False}, + "local_rpms_dir": {"type": "array", "items": {"type": "string"}}, + "timeout": _timeout_prop(3600), + }, + ["project_path"], + ), + "annotations": _mutating_annotations("Build Sailfish RPM"), + } + + +def _spec_obs_results() -> dict[str, Any]: + return { + "name": "sailfish_obs_results", + "title": "OBS Results", + "description": "Run osc results using the configured OBS API alias when set.", + "inputSchema": _object_schema( + { + "project": {"type": "string"}, + "package": {"type": "string"}, + "api_alias": {"type": "string"}, + "timeout": _timeout_prop(60), + }, + ["project"], + ), + "annotations": _read_only_annotations("OBS Results"), + } + + +def _spec_obs_buildlog() -> dict[str, Any]: + return { + "name": "sailfish_obs_buildlog", + "title": "OBS Build Log", + "description": "Fetch an OBS remote build log with osc.", + "inputSchema": _object_schema( + { + "project": {"type": "string"}, + "package": {"type": "string"}, + "repository": {"type": "string"}, + "arch": {"type": "string"}, + "api_alias": {"type": "string"}, + "timeout": _timeout_prop(90), + }, + ["project", "package", "repository", "arch"], + ), + "annotations": _read_only_annotations("OBS Build Log"), + } + + +def _spec_repo_status() -> dict[str, Any]: + return { + "name": "sailfish_repo_status", + "title": "Repo Status", + "description": "Run git status --short --branch under the configured git root.", + "inputSchema": _object_schema({"path": {"type": "string"}, "timeout": _timeout_prop(30)}), + "annotations": _read_only_annotations("Repo Status"), + } + + +def _spec_repo_find() -> dict[str, Any]: + return { + "name": "sailfish_repo_find", + "title": "Repo Find", + "description": "Search a repo or subtree under the configured git root.", + "inputSchema": _object_schema( + { + "path": {"type": "string"}, + "query": {"type": "string"}, + "fixed_strings": {"type": "boolean", "default": True}, + "max_count": {"type": "integer", "minimum": 1, "maximum": 1000, "default": 100}, + "timeout": _timeout_prop(30), + }, + ["query"], + ), + "annotations": _read_only_annotations("Repo Find"), + } + + +def _spec_spec_summary() -> dict[str, Any]: + return { + "name": "sailfish_spec_summary", + "title": "RPM Spec Summary", + "description": "Parse high-level metadata from a Sailfish RPM spec file.", + "inputSchema": _object_schema( + {"repo_path": {"type": "string"}, "spec_path": {"type": "string"}} + ), + "annotations": _read_only_annotations("RPM Spec Summary"), + } + + +def _spec_qml_find_translations() -> dict[str, Any]: + return { + "name": "sailfish_qml_find_translations", + "title": "Find QML Translations", + "description": "Find qsTrId and translator comments in QML files.", + "inputSchema": _object_schema({"path": {"type": "string"}, "timeout": _timeout_prop(30)}), + "annotations": _read_only_annotations("Find QML Translations"), + } + + +def _spec_qml_check_translator_ternaries() -> dict[str, Any]: + return { + "name": "sailfish_qml_check_translator_ternaries", + "title": "Check QML Ternary Translations", + "description": "Flag QML ternary qsTrId expressions that need branch-local comments.", + "inputSchema": _object_schema({"path": {"type": "string"}}), + "annotations": _read_only_annotations("Check QML Ternary Translations"), + } diff --git a/src/sailfish_devel_mcp/vendor/__init__.py b/src/sailfish_devel_mcp/vendor/__init__.py new file mode 100644 index 0000000..589c875 --- /dev/null +++ b/src/sailfish_devel_mcp/vendor/__init__.py @@ -0,0 +1,2 @@ +"""Vendored helper scripts used by the MCP server.""" + diff --git a/src/sailfish_devel_mcp/vendor/build_sailfishos.py b/src/sailfish_devel_mcp/vendor/build_sailfishos.py new file mode 100755 index 0000000..98b52c0 --- /dev/null +++ b/src/sailfish_devel_mcp/vendor/build_sailfishos.py @@ -0,0 +1,1459 @@ +#!/usr/bin/env python3 + +import argparse +import json +import os +import re +import shlex +import shutil +import subprocess +import sys +from dataclasses import dataclass +from datetime import datetime, timezone +from pathlib import Path +from typing import Iterable +from urllib.error import HTTPError, URLError +from urllib.parse import quote +from urllib.request import urlopen + + +CONTAINER_UID = 100000 +CONTAINER_IMAGE = "coderus/sailfishos-platform-sdk" +LIVE_RELEASE = "live" +DEFAULT_LOCAL_SDK = Path("/srv/mer/sdks/sfossdk/sdk-chroot") +LOCAL_SDK_BUILD_ENGINE_IMAGE_ENV = "SAILFISH_SDK_BUILD_ENGINE_IMAGE" +STATE_DIRNAME = "build-sailfishos-skill" +MANIFEST_NAME = "build-sailfishos-skill-manifest.txt" +BUILD_LOG_NAME = "build-sailfishos-skill-last.log" +BUILD_METADATA_NAME = "build-sailfishos-skill-last-build.json" +DEFAULT_PERMISSION_FALLBACK = "error" + +ROOT_PATTERNS = ( + "Makefile", + ".qmake.stash", + "*.o", + "*.a", + "*.so", + "*.prl", + "moc_*.cpp", + "moc_*.o", + "qrc_*.cpp", + "qrc_*.o", + "ui_*.h", + "CMakeCache.txt", + "cmake_install.cmake", + "compile_commands.json", + "build.ninja", + "rules.ninja", + "install_manifest.txt", +) + +RECURSIVE_PATTERNS = ( + "**/Makefile", + "**/.qmake.stash", + "**/moc_*.cpp", + "**/moc_*.o", + "**/qrc_*.cpp", + "**/qrc_*.o", + "**/*.o", + "**/*.a", + "**/*.so", + "**/*.prl", + "**/ui_*.h", + "**/CMakeCache.txt", + "**/cmake_install.cmake", + "**/compile_commands.json", + "**/build.ninja", + "**/rules.ninja", + "**/install_manifest.txt", + "**/CMakeFiles", +) + +ROOT_DIRS = ( + "installroot", +) + +LOCAL_TARGET_ARCHES = ("aarch64", "armv7hl", "i486") + + +@dataclass(frozen=True) +class LocalSdkTarget: + arch: str + target: str + release: str + version_id: str + flavour: str + + +@dataclass(frozen=True) +class LocalSdkBuild: + arch: str + target: str + + +def log(message: str) -> None: + print(message, file=sys.stderr) + + +def run(cmd: list[str], cwd: Path | None = None, capture_output: bool = False) -> subprocess.CompletedProcess[str]: + return subprocess.run( + cmd, + cwd=str(cwd) if cwd else None, + check=True, + text=True, + capture_output=capture_output, + ) + + +def require_tool(name: str) -> None: + if shutil.which(name): + return + raise SystemExit(f"Required tool not found: {name}") + + +def project_state_dir(project_dir: Path) -> Path: + return project_dir / ".mb2" / STATE_DIRNAME + + +def manifest_path(project_dir: Path) -> Path: + return project_dir / ".mb2" / MANIFEST_NAME + + +def build_log_path(project_dir: Path) -> Path: + return project_dir / ".mb2" / BUILD_LOG_NAME + + +def build_metadata_path(project_dir: Path) -> Path: + return project_dir / ".mb2" / BUILD_METADATA_NAME + + +def default_artifacts_dir(project_dir: Path) -> Path: + return project_dir / "RPMS" + + +def staging_rpms_dir(project_dir: Path) -> Path: + return project_state_dir(project_dir) / "rpms" + + +def has_spec_files(project_dir: Path) -> bool: + rpm_dir = project_dir / "rpm" + return rpm_dir.is_dir() and any(rpm_dir.glob("*.spec")) + + +def parse_version(value: str) -> tuple[int, ...]: + return tuple(int(part) for part in value.split(".")) + + +def is_version_release_tag(value: str) -> bool: + return bool(re.fullmatch(r"\d+(?:\.\d+){3}", value)) + + +def fetch_release_tags(prefix: str | None = None) -> list[str]: + name_filter = quote(prefix) if prefix else "" + + matches: list[str] = [] + next_url = ( + "https://registry.hub.docker.com/v2/repositories/" + f"{CONTAINER_IMAGE}/tags?page_size=100" + f"{f'&name={name_filter}' if name_filter else ''}" + ) + while next_url: + with urlopen(next_url, timeout=10) as response: + payload = json.load(response) + for result in payload.get("results", []): + name = result.get("name", "").strip() + if prefix: + if not (name == prefix or name.startswith(f"{prefix}.")): + continue + if is_version_release_tag(name): + matches.append(name) + next_url = payload.get("next") + return sorted(dict.fromkeys(matches), key=parse_version) + + +def latest_release_tag() -> str: + matches = fetch_release_tags() + if not matches: + raise SystemExit( + f"Could not determine the latest SailfishOS release from {CONTAINER_IMAGE} tags." + ) + return matches[-1] + + +def normalize_release_tag(release: str) -> str: + if release.lower() == LIVE_RELEASE: + return LIVE_RELEASE + + if release == "latest": + try: + resolved = latest_release_tag() + except (HTTPError, URLError, TimeoutError, OSError, ValueError, json.JSONDecodeError) as exc: + raise SystemExit("Could not resolve SailfishOS release tag 'latest'") from exc + log(f"Resolved SailfishOS release latest to {resolved}") + return resolved + + if not re.fullmatch(r"\d+(?:\.\d+){2,3}", release): + return release + if release.count(".") >= 3: + return release + + try: + matches = fetch_release_tags(release) + except (HTTPError, URLError, TimeoutError, OSError, ValueError, json.JSONDecodeError): + return release + + if not matches: + return release + if release in matches: + return release + + resolved = max(matches, key=parse_version) + log(f"Resolved SailfishOS release {release} to {resolved}") + return resolved + + +def infer_release_from_workflows(project_dir: Path) -> str | None: + workflows_dir = project_dir / ".github" / "workflows" + if not workflows_dir.is_dir(): + return None + + regexes = ( + re.compile(r"^\s*RELEASE:\s*([^\s#]+)\s*$"), + re.compile(rf"{re.escape(CONTAINER_IMAGE)}:([^\s'\"#]+)"), + ) + + for workflow in sorted(workflows_dir.glob("*.y*ml")): + try: + text = workflow.read_text(encoding="utf-8") + except OSError: + continue + for line in text.splitlines(): + for regex in regexes: + match = regex.search(line) + if match: + return match.group(1).strip() + return None + + +def resolve_release(project_dirs: Iterable[Path], explicit_release: str | None) -> str: + if explicit_release: + return normalize_release_tag(explicit_release) + + env_release = os.environ.get("SAILFISHOS_RELEASE") + if env_release: + return normalize_release_tag(env_release) + + seen: set[Path] = set() + for project_dir in project_dirs: + if project_dir in seen: + continue + seen.add(project_dir) + inferred = infer_release_from_workflows(project_dir) + if inferred: + return normalize_release_tag(inferred) + + try: + resolved = latest_release_tag() + except (HTTPError, URLError, TimeoutError, OSError, ValueError, json.JSONDecodeError) as exc: + raise SystemExit( + "Could not determine SailfishOS release from arguments, environment, workflows, or Docker tags." + ) from exc + + log(f"No SailfishOS release specified; using latest available release {resolved}") + return resolved + + +def parse_last_arch(project_dir: Path) -> str | None: + target_file = project_dir / ".mb2" / "target" + if not target_file.is_file(): + return None + + target = target_file.read_text(encoding="utf-8").strip() + if not target: + return None + + if target.startswith("SailfishOS-"): + arch = target.rsplit("-", 1)[-1] + if arch.endswith(".default"): + arch = arch[: -len(".default")] + return arch + + prefix = target.split(".", 1)[0].strip() + return prefix or None + + +def pull_image(release: str) -> None: + image = f"{CONTAINER_IMAGE}:{release}" + log(f"Pulling {image}") + run(["docker", "pull", image]) + + +def list_supported_arches(release: str) -> list[str]: + image = f"{CONTAINER_IMAGE}:{release}" + result = run( + [ + "docker", + "run", + "--rm", + image, + "bash", + "-lc", + "sb2-config -l", + ], + capture_output=True, + ) + + arches: list[str] = [] + seen: set[str] = set() + prefix = f"SailfishOS-{release}-" + for line in result.stdout.splitlines(): + line = line.strip() + if line.startswith(prefix): + arch = line[len(prefix) :] + if arch.endswith(".default") or arch in seen: + continue + arches.append(arch) + seen.add(arch) + if not arches: + raise SystemExit(f"No supported architectures found in {image}") + return arches + + +def resolve_arches(requested_arches: list[str], build_all: bool, supported_arches: list[str], project_dir: Path) -> list[str]: + if build_all: + return supported_arches + + if requested_arches: + invalid = [arch for arch in requested_arches if arch not in supported_arches] + if invalid: + raise SystemExit( + f"Unsupported architectures: {', '.join(invalid)}. Supported: {', '.join(supported_arches)}" + ) + return requested_arches + + last_arch = parse_last_arch(project_dir) + if last_arch and last_arch in supported_arches: + return [last_arch] + + raise SystemExit( + "No architecture was specified and .mb2/target did not contain a supported one. " + f"Pass --arch or --all. Supported: {', '.join(supported_arches)}" + ) + + +def resolve_local_sdk_arches(requested_arches: list[str], build_all: bool, project_dir: Path) -> list[str]: + if build_all: + raise SystemExit("Local SDK builds use installed SDK targets; pass one or more explicit --arch values.") + + if requested_arches: + return requested_arches + + last_arch = parse_last_arch(project_dir) + if last_arch: + return [last_arch] + + raise SystemExit( + "No architecture was specified and .mb2/target did not contain a previous target. " + "Pass --arch for local SDK builds." + ) + + +def spec_names(project_dir: Path) -> set[str]: + names: set[str] = set() + for spec in sorted((project_dir / "rpm").glob("*.spec")): + try: + text = spec.read_text(encoding="utf-8") + except OSError: + continue + for line in text.splitlines(): + match = re.match(r"^\s*Name:\s*(\S+)\s*$", line) + if match: + names.add(match.group(1)) + break + return names + + +def pro_targets(project_dir: Path) -> set[str]: + targets: set[str] = set() + for pro in sorted(project_dir.glob("*.pro")): + try: + text = pro.read_text(encoding="utf-8") + except OSError: + continue + for line in text.splitlines(): + match = re.match(r"^\s*TARGET\s*=\s*([^\s#]+)\s*$", line) + if match and "$" not in match.group(1): + targets.add(match.group(1)) + break + return targets + + +def generated_candidate_paths(project_dir: Path) -> set[Path]: + tracked_paths = tracked_git_paths(project_dir) + paths: set[Path] = set() + + for dirname in ROOT_DIRS: + path = project_dir / dirname + if path.exists(): + paths.add(path) + + for pattern in ROOT_PATTERNS: + paths.update(path for path in project_dir.glob(pattern) if path.exists()) + + for pattern in RECURSIVE_PATTERNS: + paths.update( + path + for path in project_dir.glob(pattern) + if path.exists() and ".git" not in path.parts and ".mb2" not in path.parts + ) + + for qm in (project_dir / "translations").glob("*.qm") if (project_dir / "translations").is_dir() else []: + if qm.exists(): + paths.add(qm) + + for name in sorted(spec_names(project_dir) | pro_targets(project_dir)): + candidate = project_dir / name + if candidate.exists(): + paths.add(candidate) + + state_dir = project_state_dir(project_dir) + if state_dir.exists(): + paths.discard(state_dir) + + return { + path + for path in paths + if path.exists() and path.resolve() not in tracked_paths + } + + +def tracked_git_paths(project_dir: Path) -> set[Path]: + try: + repo_roots = git_worktree_roots(project_dir) + except OSError: + return set() + + tracked: set[Path] = set() + for root in repo_roots: + try: + result = run( + ["git", "-C", str(root), "ls-files", "-z"], + capture_output=True, + ) + except (subprocess.CalledProcessError, FileNotFoundError): + continue + + for rel_path in result.stdout.split("\0"): + if not rel_path: + continue + tracked.add((root / rel_path).resolve()) + return tracked + + +def git_worktree_roots(project_dir: Path) -> list[Path]: + roots = {project_dir.resolve()} + for git_marker in project_dir.rglob(".git"): + if ".mb2" in git_marker.parts: + continue + repo_root = git_marker.parent.resolve() + roots.add(repo_root) + return sorted(roots) + + +def load_manifest(project_dir: Path) -> list[Path]: + path = manifest_path(project_dir) + if not path.is_file(): + return [] + + result: list[Path] = [] + for line in path.read_text(encoding="utf-8").splitlines(): + line = line.strip() + if not line: + continue + candidate = (project_dir / line).resolve() + try: + candidate.relative_to(project_dir.resolve()) + except ValueError: + continue + if candidate.exists(): + result.append(candidate) + return result + + +def write_manifest(project_dir: Path, paths: Iterable[Path]) -> None: + manifest = manifest_path(project_dir) + manifest.parent.mkdir(parents=True, exist_ok=True) + + rel_paths = [] + root = project_dir.resolve() + for path in sorted({p.resolve() for p in paths if p.exists()}): + try: + rel_paths.append(str(path.relative_to(root))) + except ValueError: + continue + + manifest.write_text("\n".join(rel_paths) + ("\n" if rel_paths else ""), encoding="utf-8") + + +def write_build_metadata( + project_dir: Path, + *, + release: str, + arch: str, + debug_build: bool, + artifacts_dir: Path, + status: str, + rpms: Iterable[Path], +) -> None: + metadata_file = build_metadata_path(project_dir) + metadata_file.parent.mkdir(parents=True, exist_ok=True) + payload = { + "timestamp_utc": datetime.now(timezone.utc).isoformat(), + "release": release, + "arch": arch, + "debug": debug_build, + "status": status, + "artifacts_dir": str(artifacts_dir), + "build_log": str(build_log_path(project_dir)), + "rpms": [str(path) for path in rpms], + } + metadata_file.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8") + + +def write_target_marker(project_dir: Path, arch: str) -> None: + mb2_dir = project_dir / ".mb2" + mb2_dir.mkdir(parents=True, exist_ok=True) + (mb2_dir / "target").write_text(f"{arch}.{STATE_DIRNAME}\n", encoding="utf-8") + + +def remove_path(path: Path) -> None: + if not path.exists(): + return + if path.is_dir() and not path.is_symlink(): + shutil.rmtree(path) + else: + path.unlink() + + +def cleanup_generated_artifacts(project_dir: Path, reason: str) -> list[Path]: + manifest_paths = load_manifest(project_dir) + cleanup_paths = set(manifest_paths) | generated_candidate_paths(project_dir) + + removed: list[Path] = [] + for path in sorted(cleanup_paths): + if project_state_dir(project_dir) in path.parents or path == project_state_dir(project_dir): + continue + if path == manifest_path(project_dir): + continue + if path.exists(): + remove_path(path) + removed.append(path) + + log(f"{reason}; removed {len(removed)} stale in-place artifacts") + return removed + + +def cleanup_in_place_artifacts(project_dir: Path, previous_arch: str, next_arch: str) -> list[Path]: + return cleanup_generated_artifacts( + project_dir, + f"Switched architecture from {previous_arch} to {next_arch}", + ) + + +def ensure_container_write_access(project_dir: Path, permission_fallback: str) -> None: + current_uid = os.getuid() + if shutil.which("setfacl"): + log( + f"Granting write ACLs to host uid {current_uid} and container uid {CONTAINER_UID} under {project_dir}" + ) + run( + [ + "find", + str(project_dir), + "(", + "-type", + "f", + "-o", + "-type", + "d", + ")", + "-uid", + str(current_uid), + "-exec", + "setfacl", + "-m", + f"u:{current_uid}:rwX,u:{CONTAINER_UID}:rwX", + "{}", + "+", + ] + ) + run( + [ + "find", + str(project_dir), + "-type", + "d", + "-uid", + str(current_uid), + "-exec", + "setfacl", + "-m", + f"d:u:{current_uid}:rwX,d:u:{CONTAINER_UID}:rwX", + "{}", + "+", + ] + ) + return + + if permission_fallback == "chmod": + log("setfacl unavailable; falling back to chmod -R a+rwX") + run(["chmod", "-R", "a+rwX", str(project_dir)]) + return + + raise SystemExit( + "setfacl is unavailable, so the Docker container may not be able to write in place. " + "Install acl utilities or rerun with --permission-fallback chmod." + ) + + +def parse_missing_build_requires(log_text: str) -> list[str]: + missing: list[str] = [] + capture = False + for line in log_text.splitlines(): + if line.strip() == "error: Failed build dependencies:": + capture = True + continue + if not capture: + continue + if line.startswith("\t") or line.startswith(" "): + requirement = line.strip() + if " is needed by " in requirement: + requirement = requirement.split(" is needed by ", 1)[0].strip() + if requirement: + missing.append(requirement) + continue + if missing and line.strip(): + break + return sorted(dict.fromkeys(missing)) + + +def extract_zypper_names(output: str) -> list[str]: + names: list[str] = [] + for line in output.splitlines(): + if "|" not in line or line.lstrip().startswith("--+"): + continue + parts = [part.strip() for part in line.split("|")] + if len(parts) < 3: + continue + name = parts[1] + if name and name not in {"Name", "S"}: + names.append(name) + return sorted(dict.fromkeys(names)) + + +def diagnose_missing_dependencies(project_dir: Path, release: str, arch: str) -> None: + log_file = build_log_path(project_dir) + if not log_file.is_file(): + return + + missing = parse_missing_build_requires(log_file.read_text(encoding="utf-8")) + if not missing: + return + + image = f"{CONTAINER_IMAGE}:{release}" + target = f"SailfishOS-{release}-{arch}" + log("Dependency diagnostics from the target SDK:") + for requirement in missing: + if requirement.startswith("pkgconfig("): + query = ( + f"sb2 -t {shlex.quote(target)} -m sdk-install -R " + f"zypper search --provides --match-exact {shlex.quote(requirement)}" + ) + else: + query = ( + f"sb2 -t {shlex.quote(target)} -m sdk-install -R " + f"zypper se -s {shlex.quote(requirement)}" + ) + + try: + result = run( + ["docker", "run", "--rm", image, "bash", "-lc", query], + capture_output=True, + ) + except subprocess.CalledProcessError: + log(f"- {requirement}: diagnostic lookup failed") + continue + + names = extract_zypper_names(result.stdout) + if names: + log(f"- {requirement}: available as {', '.join(names)}") + else: + log(f"- {requirement}: not available in {target}") + + +def verify_expected_rpms(rpms: list[Path], debug_build: bool) -> None: + if not rpms: + raise SystemExit("Build completed but no RPMs were captured") + if debug_build: + names = {rpm.name for rpm in rpms} + if not any("-debuginfo-" in name for name in names): + raise SystemExit("Debug build completed but no -debuginfo RPM was produced") + if not any("-debugsource-" in name for name in names): + raise SystemExit("Debug build completed but no -debugsource RPM was produced") + + +def variant_destination_dir(base_dir: Path, release: str, arch: str, debug_build: bool) -> Path: + return base_dir / release / arch / ("debug" if debug_build else "release") + + +def host_user() -> str: + return os.environ.get("USER") or os.environ.get("LOGNAME") or Path.home().name + + +def local_sdk_build_engine_image(user: str) -> str: + return os.environ.get(LOCAL_SDK_BUILD_ENGINE_IMAGE_ENV, f"sailfish-sdk-build-engine:{user}") + + +def local_sdk_project_mount_root(project_dir: Path) -> Path: + home = Path.home().resolve() + resolved = project_dir.resolve() + try: + resolved.relative_to(home) + except ValueError as exc: + raise SystemExit( + "Local SDK builds currently require the project to live under the current user's home " + "directory so the installed SDK chroot can see the same path." + ) from exc + return home + + +def local_sdk_mount_root(local_sdk: Path) -> Path: + resolved = local_sdk.resolve(strict=False) + srv_mer = Path("/srv/mer") + try: + resolved.relative_to(srv_mer) + except ValueError: + return resolved.parent + return srv_mer + + +def local_sdk_targets_dir(local_sdk: Path) -> Path: + return local_sdk_mount_root(local_sdk) / "targets" + + +def canonical_local_target_name(name: str) -> str: + while name.endswith(".default"): + name = name[: -len(".default")] + return name + + +def split_local_target_arch(target: str) -> tuple[str, str] | None: + for arch in LOCAL_TARGET_ARCHES: + if target == arch: + return arch, "" + if target.startswith(f"{arch}-"): + return arch, target[len(arch) + 1 :] + return None + + +def read_key_value_file(path: Path) -> dict[str, str]: + if not path.is_file(): + return {} + + metadata: dict[str, str] = {} + for line in path.read_text(encoding="utf-8").splitlines(): + line = line.strip() + if not line or line.startswith("[") or line.startswith("#"): + continue + if "=" not in line: + continue + key, value = line.split("=", 1) + metadata[key.strip()] = value.strip().strip('"') + return metadata + + +def target_metadata(target_dir: Path) -> dict[str, str]: + sailfish = read_key_value_file(target_dir / "etc" / "sailfish-release") + ssu = read_key_value_file(target_dir / "etc" / "ssu" / "ssu.ini") + return { + "release": ssu.get("release", ""), + "version_id": sailfish.get("VERSION_ID", ""), + "flavour": ssu.get("flavour") or sailfish.get("SAILFISH_FLAVOUR", ""), + } + + +def list_local_sdk_targets(local_sdk: Path) -> list[LocalSdkTarget]: + targets_dir = local_sdk_targets_dir(local_sdk) + if not targets_dir.is_dir(): + return [] + + targets: dict[str, LocalSdkTarget] = {} + for child in sorted(targets_dir.iterdir()): + if not child.is_dir() or ".pool." in child.name: + continue + + target = canonical_local_target_name(child.name) + arch_and_suffix = split_local_target_arch(target) + if arch_and_suffix is None: + continue + arch, suffix = arch_and_suffix + + if target in targets and child.name != target: + continue + + metadata = target_metadata(child) + release = metadata.get("release", "") + version_id = metadata.get("version_id", "") + flavour = metadata.get("flavour", "") + if not release and suffix: + release = suffix + targets[target] = LocalSdkTarget( + arch=arch, + target=target, + release=release, + version_id=version_id, + flavour=flavour, + ) + return sorted(targets.values(), key=lambda item: (item.arch, item.target)) + + +def release_component_count(release: str) -> int: + return len(release.split(".")) if re.fullmatch(r"\d+(?:\.\d+){2,3}", release) else 0 + + +def local_target_matches_release(target: LocalSdkTarget, release: str) -> bool: + release = normalize_local_release(release) + if not release or release == LIVE_RELEASE: + return target.release == LIVE_RELEASE + if release == "latest": + return False + + if target.release == release or target.version_id == release: + return True + + component_count = release_component_count(release) + if component_count == 3: + return target.version_id.startswith(f"{release}.") or target.target.endswith(f"-{release}") + + return target.target.endswith(f"-{release}") + + +def normalize_local_release(release: str | None) -> str: + if not release: + return "" + release = release.strip() + if release.lower() == LIVE_RELEASE: + return LIVE_RELEASE + return release + + +def requested_release(project_dirs: Iterable[Path], explicit_release: str | None) -> str | None: + if explicit_release: + return explicit_release + + env_release = os.environ.get("SAILFISHOS_RELEASE") + if env_release: + return env_release + + seen: set[Path] = set() + for project_dir in project_dirs: + if project_dir in seen: + continue + seen.add(project_dir) + inferred = infer_release_from_workflows(project_dir) + if inferred: + return inferred + + return None + + +def select_local_sdk_builds( + local_sdk: Path, + release: str, + requested_arches: list[str], + build_all: bool, + project_dir: Path, +) -> list[LocalSdkBuild] | None: + matching = [ + target + for target in list_local_sdk_targets(local_sdk) + if local_target_matches_release(target, release) + ] + if not matching: + return None + + by_arch = {target.arch: target for target in matching} + by_target = {target.target: target for target in matching} + + if build_all: + return [LocalSdkBuild(target.arch, target.target) for target in matching] + + requested = requested_arches[:] + if not requested: + last_arch = parse_last_arch(project_dir) + if last_arch: + requested = [last_arch] + + if not requested: + return None + + builds: list[LocalSdkBuild] = [] + for arch in requested: + target = by_target.get(arch) or by_arch.get(arch) + if target is None: + return None + builds.append(LocalSdkBuild(target.arch, target.target)) + return builds + + +def build_local_sdk_arch( + project_dir: Path, + local_sdk: Path, + release: str, + arch: str, + target: str, + debug_build: bool = False, + local_rpm_dirs: list[Path] | None = None, +) -> None: + user = host_user() + uid = os.getuid() + gid = os.getgid() + home = str(Path.home().resolve()) + image = local_sdk_build_engine_image(user) + project_mount_root = local_sdk_project_mount_root(project_dir) + sdk_mount_root = local_sdk_mount_root(local_sdk) + binary_names = ":".join(sorted(spec_names(project_dir) | pro_targets(project_dir))) + local_rpm_dirs = local_rpm_dirs or [] + + inner_command = r''' +set -euo pipefail +cd "$PROJECT_DIR" +mkdir -p .mb2 +mkdir -p .mb2/build-sailfishos-skill +logfile="$BUILD_LOG" +: > "$logfile" +{ + echo "# build-sailfishos-skill" + echo "release=${RELEASE:-}" + echo "arch=${ARCH:-}" + echo "debug=${DEBUG_BUILD:-0}" + echo "target=${TARGET:-}" + echo +} >> "$logfile" + +if [ -n "${LOCAL_RPM_DIRS:-}" ]; then + rpm_files=() + OLDIFS="$IFS" + IFS=':' + for dir in ${LOCAL_RPM_DIRS}; do + [ -d "$dir" ] || continue + for rpm in "$dir"/*.rpm; do + [ -e "$rpm" ] || continue + case "$(basename "$rpm")" in + *-debuginfo-*|*-debugsource-*|*-tests-*|*-examples-*|*-doc-*|*-ts-devel-*) + continue + ;; + esac + rpm_files+=("$rpm") + done + done + IFS="$OLDIFS" + if [ "${#rpm_files[@]}" -gt 0 ]; then + sb2 -t "$TARGET" -m sdk-install -R zypper --non-interactive install \ + --allow-unsigned-rpm --oldpackage --force-resolution "${rpm_files[@]}" 2>&1 | tee -a "$logfile" + fi +fi + +mb2_args=( -t "$TARGET" --no-vcs-apply build --prepare ) +if [ "${DEBUG_BUILD:-0}" = "1" ]; then + mb2_args+=( -d ) +fi +mb2 "${mb2_args[@]}" 2>&1 | tee -a "$logfile" + +rm -rf .mb2/build-sailfishos-skill/rpms +if [ -d RPMS ]; then + mkdir -p .mb2/build-sailfishos-skill/rpms + find RPMS -maxdepth 1 -type f -name '*.rpm' -exec cp -f {} .mb2/build-sailfishos-skill/rpms/ \; + chmod -R u+rwX .mb2/build-sailfishos-skill/rpms >/dev/null 2>&1 || true +fi + +OLDIFS="$IFS" +IFS=':' +for name in ${SYNC_BINARIES:-}; do + [ -n "$name" ] || continue + [ -e "$name" ] || continue + cp -f "$name" .mb2/build-sailfishos-skill/ >/dev/null 2>&1 || true +done +IFS="$OLDIFS" +''' + wrapper_command = rf''' +set -euo pipefail +if [ ! -x "$LOCAL_SDK" ]; then + echo "Installed Sailfish SDK chroot not found or not executable at $LOCAL_SDK" >&2 + exit 1 +fi +if getent passwd mersdk >/dev/null 2>&1; then + sed -i 's#^mersdk:[^:]*:[0-9]*:[0-9]*:[^:]*:[^:]*:#{user}:x:{uid}:{gid}::{home}:#' /etc/passwd +elif ! getent passwd {shlex.quote(user)} >/dev/null 2>&1; then + printf '%s:x:%s:%s::%s:/bin/bash\n' {shlex.quote(user)} {uid} {gid} {shlex.quote(home)} >> /etc/passwd +fi +"$LOCAL_SDK" -u {shlex.quote(user)} env \ + PROJECT_DIR="$PROJECT_DIR" \ + RELEASE="$RELEASE" \ + TARGET="$TARGET" \ + ARCH="$ARCH" \ + DEBUG_BUILD="$DEBUG_BUILD" \ + BUILD_LOG="$BUILD_LOG" \ + LOCAL_RPM_DIRS="$LOCAL_RPM_DIRS" \ + SYNC_BINARIES="$SYNC_BINARIES" \ + bash -lc {shlex.quote(inner_command)} +''' + log(f"Building local SDK target {target} for {release} via installed /srv/mer SDK") + run( + [ + "docker", + "run", + "--rm", + "--privileged", + "-v", + f"{sdk_mount_root}:{sdk_mount_root}", + "-v", + f"{project_mount_root}:{project_mount_root}", + "-w", + str(project_dir), + "-e", + f"PROJECT_DIR={project_dir}", + "-e", + f"LOCAL_SDK={local_sdk}", + "-e", + f"RELEASE={release}", + "-e", + f"TARGET={target}", + "-e", + f"ARCH={arch}", + "-e", + f"DEBUG_BUILD={'1' if debug_build else '0'}", + "-e", + f"BUILD_LOG={build_log_path(project_dir)}", + "-e", + f"LOCAL_RPM_DIRS={':'.join(str(path) for path in local_rpm_dirs)}", + "-e", + f"SYNC_BINARIES={binary_names}", + image, + "bash", + "-lc", + wrapper_command, + ] + ) + + +def build_arch( + project_dir: Path, + release: str, + arch: str, + debug_build: bool = False, + local_rpm_dirs: list[Path] | None = None, +) -> None: + image = f"{CONTAINER_IMAGE}:{release}" + target = f"SailfishOS-{release}-{arch}" + binary_names = ":".join(sorted(spec_names(project_dir) | pro_targets(project_dir))) + is_gecko_build = (project_dir / "gecko-dev").is_dir() and (project_dir / "rpm" / "xulrunner-qt5.spec").is_file() + local_rpm_dirs = local_rpm_dirs or [] + local_rpm_mounts = [f"/local-rpms/{index}" for index, _ in enumerate(local_rpm_dirs)] + build_command = r''' +set -euo pipefail +workroot="${HOME:-/tmp}" +if [ ! -d "$workroot" ] || [ ! -w "$workroot" ]; then + workroot=/tmp +fi +workdir="$workroot/build-sailfishos-skill" +mkdir -p /share/.mb2 +mkdir -p /share/.mb2/build-sailfishos-skill +logfile=/share/.mb2/build-sailfishos-skill-last.log +: > "$logfile" +{ + echo "# build-sailfishos-skill" + echo "release=${RELEASE:-}" + echo "arch=${ARCH:-}" + echo "debug=${DEBUG_BUILD:-0}" + echo "target=${TARGET:-}" + echo +} >> "$logfile" +rm -rf "$workdir" +mkdir -p "$workdir" +cp -a /share/. "$workdir/" +rm -rf "$workdir/RPMS" +cd "$workdir" +if [ "${IS_GECKO_BUILD:-0}" = "1" ]; then + # The local Sailfish gecko checkout already has the rpm/ patch stack applied, + # so keep %prep for its bootstrap side effects but disable patch re-apply in + # the copied spec. + sed -i \ + -e 's/^%autosetup -p1 -n /%autosetup -N -n /' \ + rpm/xulrunner-qt5.spec +fi + +if [ -n "${LOCAL_RPM_DIRS:-}" ]; then + rpm_files=() + OLDIFS="$IFS" + IFS=':' + for dir in ${LOCAL_RPM_DIRS}; do + [ -d "$dir" ] || continue + for rpm in "$dir"/*.rpm; do + [ -e "$rpm" ] || continue + case "$(basename "$rpm")" in + *-debuginfo-*|*-debugsource-*|*-tests-*|*-examples-*|*-doc-*|*-ts-devel-*) + continue + ;; + esac + rpm_files+=("$rpm") + done + done + IFS="$OLDIFS" + if [ "${#rpm_files[@]}" -gt 0 ]; then + zypper --non-interactive install --allow-unsigned-rpm --oldpackage --force-resolution \ + "${rpm_files[@]}" 2>&1 | tee -a "$logfile" + fi +fi + +mb2_args=( -t "$TARGET" ) +if [ "${IS_GECKO_BUILD:-0}" = "1" ]; then + mb2_args+=( --no-vcs-apply ) +fi +mb2_args+=( build ) +if [ "${IS_GECKO_BUILD:-0}" = "1" ]; then + mb2_args+=( --prepare ) +fi +if [ "${DEBUG_BUILD:-0}" = "1" ]; then + mb2_args+=( -d ) +fi +mb2 "${mb2_args[@]}" 2>&1 | tee -a "$logfile" + +for state_file in .mb2/target .mb2/spec .mb2/snapshot.lock; do + if [ -e "$state_file" ]; then + cp -f "$state_file" /share/.mb2/ + fi +done +chmod -R a+rwX /share/.mb2 >/dev/null 2>&1 || true + +rm -rf /share/.mb2/build-sailfishos-skill/rpms +if [ -d RPMS ]; then + mkdir -p /share/.mb2/build-sailfishos-skill/rpms + find RPMS -maxdepth 1 -type f -name '*.rpm' -exec cp -f {} /share/.mb2/build-sailfishos-skill/rpms/ \; + chmod -R a+rwX /share/.mb2/build-sailfishos-skill/rpms >/dev/null 2>&1 || true +fi + +for pattern in \ + Makefile .qmake.stash '*.o' '*.a' '*.so' '*.prl' '*.list' \ + 'moc_*.cpp' 'moc_*.o' 'qrc_*.cpp' 'qrc_*.o' 'ui_*.h' \ + CMakeCache.txt cmake_install.cmake compile_commands.json build.ninja rules.ninja install_manifest.txt +do + for f in $pattern; do + [ -e "$f" ] || continue + cp -f "$f" /share/ + done +done + +if [ -d translations ]; then + mkdir -p /share/translations + for f in translations/*.qm; do + [ -e "$f" ] || continue + cp -f "$f" /share/translations/ + done +fi + +OLDIFS="$IFS" +IFS=':' +for name in ${SYNC_BINARIES:-}; do + [ -n "$name" ] || continue + [ -e "$name" ] || continue + cp -f "$name" /share/ +done +IFS="$OLDIFS" +''' + gecko_wrapper_command = rf''' +set -euo pipefail +if [ ! -e /usr/lib/libclang.so.15 ]; then + zypper --non-interactive install clang-libs +fi +if ! rpm -q gcc-c++ >/dev/null 2>&1; then + zypper --non-interactive install gcc-c++ +fi +python3 - <<'PY' +import os +import pwd + +pw = pwd.getpwnam("mersdk") +os.environ["HOME"] = pw.pw_dir +os.setgroups([]) +os.setgid(pw.pw_gid) +os.setuid(pw.pw_uid) +os.execvp("bash", ["bash", "-lc", {build_command!r}]) +PY +''' + log(f"Building {target} via container shadow build and syncing artifacts back in place") + try: + docker_cmd = [ + "docker", + "run", + "--rm", + "--privileged", + "-v", + f"{project_dir}:/share", + ] + if is_gecko_build: + docker_cmd.extend(["-u", "0"]) + for mount_path, local_rpm_dir in zip(local_rpm_mounts, local_rpm_dirs): + docker_cmd.extend(["-v", f"{local_rpm_dir}:{mount_path}:ro"]) + docker_cmd.extend( + [ + "-e", + f"TARGET={target}", + "-e", + f"RELEASE={release}", + "-e", + f"ARCH={arch}", + "-e", + f"DEBUG_BUILD={'1' if debug_build else '0'}", + "-e", + f"BUILD_LOG={build_log_path(project_dir)}", + "-e", + f"SYNC_BINARIES={binary_names}", + "-e", + f"LOCAL_RPM_DIRS={':'.join(local_rpm_mounts)}", + "-e", + f"IS_GECKO_BUILD={'1' if is_gecko_build else '0'}", + image, + "bash", + "-lc", + gecko_wrapper_command if is_gecko_build else build_command, + ] + ) + run(docker_cmd) + except subprocess.CalledProcessError: + diagnose_missing_dependencies(project_dir, release, arch) + raise + + +def copy_rpms(project_dir: Path, release: str, arch: str, debug_build: bool, artifacts_dir: Path) -> list[Path]: + rpm_dir = staging_rpms_dir(project_dir) + if not rpm_dir.is_dir(): + raise SystemExit("Build completed but staged RPMs were not captured") + + rpms = sorted(rpm_dir.glob("*.rpm")) + if not rpms: + raise SystemExit("Build completed but no staged RPMs were found") + + destination_dir = variant_destination_dir(artifacts_dir, release, arch, debug_build) + if destination_dir.exists(): + shutil.rmtree(destination_dir) + destination_dir.mkdir(parents=True, exist_ok=True) + + copied: list[Path] = [] + for rpm in rpms: + destination = destination_dir / rpm.name + shutil.copy2(rpm, destination) + copied.append(destination) + + shutil.rmtree(rpm_dir) + return copied + + +def resolve_project_dir(project_dir: Path) -> Path: + if not project_dir.is_dir(): + raise SystemExit(f"Project directory not found: {project_dir}") + + if has_spec_files(project_dir): + return project_dir + + matches = [child for child in sorted(project_dir.iterdir()) if child.is_dir() and has_spec_files(child)] + if len(matches) == 1: + log(f"Using SailfishOS build root {matches[0]} discovered under {project_dir}") + return matches[0] + if len(matches) > 1: + options = ", ".join(str(path) for path in matches) + raise SystemExit( + f"{project_dir} contains multiple one-level-deep SailfishOS build roots: {options}. " + "Pass --project-dir pointing at the intended one." + ) + + raise SystemExit( + f"Could not find rpm/*.spec in {project_dir} or one level below it. " + "Pass --project-dir pointing at the SailfishOS build root." + ) + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description="Build a SailfishOS project in place with Docker and mb2") + parser.add_argument("--project-dir", default=".", help="Project root containing rpm/*.spec") + parser.add_argument("--release", help="SailfishOS release, for example 3.4.0.24") + parser.add_argument("--arch", action="append", default=[], help="Architecture to build, may be repeated") + parser.add_argument("--all", action="store_true", help="Build every architecture supported by the chosen SDK image") + parser.add_argument("--list-arches", action="store_true", help="Print supported architectures and exit") + parser.add_argument( + "--permission-fallback", + choices=("error", "chmod"), + default=DEFAULT_PERMISSION_FALLBACK, + help="Fallback when setfacl is unavailable", + ) + parser.add_argument( + "--artifacts-dir", + help="Directory where built RPMs are copied. Defaults to RPMS/", + ) + parser.add_argument("--clean", action="store_true", help="Remove generated in-place build artifacts before building") + parser.add_argument( + "--debug", + action="store_true", + help="Pass -d to mb2 build so main binaries are stripped and debug packages are generated", + ) + parser.add_argument( + "--local-rpms-dir", + action="append", + default=[], + help="Directory of locally built RPMs to install into the SDK target before building; may be repeated", + ) + parser.add_argument("--no-pull", action="store_true", help="Skip docker pull before building") + parser.add_argument( + "--local-sdk", + nargs="?", + const=str(DEFAULT_LOCAL_SDK), + help=( + "Use the installed SDK chroot through a privileged Docker wrapper " + f"instead of a release Docker image. Defaults to {DEFAULT_LOCAL_SDK} " + "when no path is supplied." + ), + ) + return parser.parse_args() + + +def main() -> int: + args = parse_args() + require_tool("docker") + + requested_project_dir = Path(args.project_dir).resolve() + project_dir = resolve_project_dir(requested_project_dir) + + local_sdk_path = Path(args.local_sdk).expanduser().resolve(strict=False) if args.local_sdk else None + raw_release = requested_release((requested_project_dir, project_dir), args.release) + local_release = normalize_local_release(raw_release) or LIVE_RELEASE + local_builds: list[LocalSdkBuild] | None = None + + if local_sdk_path: + local_builds = select_local_sdk_builds( + local_sdk_path, + local_release, + args.arch, + args.all or args.list_arches, + project_dir, + ) + if local_builds: + release = local_release + elif local_release == LIVE_RELEASE: + raise SystemExit("Release 'live' requires a matching installed local SDK target.") + else: + log( + f"No matching local SDK target for release {local_release}; " + f"falling back to {CONTAINER_IMAGE}" + ) + release = resolve_release((requested_project_dir, project_dir), args.release) + else: + release = resolve_release((requested_project_dir, project_dir), args.release) + if release == LIVE_RELEASE: + raise SystemExit("Release 'live' requires --local-sdk with a matching installed SDK target.") + + use_local_sdk = local_sdk_path is not None and local_builds is not None + + if not use_local_sdk and not args.no_pull: + pull_image(release) + + supported_arches = [] if use_local_sdk else list_supported_arches(release) + + if args.list_arches: + if use_local_sdk: + print("\n".join(build.arch for build in local_builds)) + return 0 + print("\n".join(supported_arches)) + return 0 + + if use_local_sdk: + builds: list[LocalSdkBuild | str] = local_builds + else: + builds = resolve_arches(args.arch, args.all, supported_arches, project_dir) + artifacts_dir = Path(args.artifacts_dir).resolve() if args.artifacts_dir else default_artifacts_dir(project_dir) + local_rpm_dirs = [Path(path).resolve() for path in args.local_rpms_dir] + + if not use_local_sdk: + ensure_container_write_access(project_dir, args.permission_fallback) + + all_copied_rpms: list[Path] = [] + for build in builds: + if isinstance(build, LocalSdkBuild): + arch = build.arch + else: + arch = build + previous_arch = parse_last_arch(project_dir) + if previous_arch and previous_arch != arch: + cleanup_in_place_artifacts(project_dir, previous_arch, arch) + elif args.clean: + cleanup_generated_artifacts(project_dir, "Explicit cleanup requested") + + try: + if isinstance(build, LocalSdkBuild): + build_local_sdk_arch( + project_dir, + local_sdk_path, + release, + arch, + build.target, + debug_build=args.debug, + local_rpm_dirs=local_rpm_dirs, + ) + else: + build_arch( + project_dir, + release, + arch, + debug_build=args.debug, + local_rpm_dirs=local_rpm_dirs, + ) + write_target_marker(project_dir, arch) + + manifest_paths = generated_candidate_paths(project_dir) + write_manifest(project_dir, manifest_paths) + + copied = copy_rpms(project_dir, release, arch, args.debug, artifacts_dir) + verify_expected_rpms(copied, args.debug) + write_build_metadata( + project_dir, + release=release, + arch=arch, + debug_build=args.debug, + artifacts_dir=artifacts_dir, + status="success", + rpms=copied, + ) + all_copied_rpms.extend(copied) + log( + f"Copied {len(copied)} RPM(s) for {arch} to " + f"{variant_destination_dir(artifacts_dir, release, arch, args.debug)}" + ) + except Exception: + write_build_metadata( + project_dir, + release=release, + arch=arch, + debug_build=args.debug, + artifacts_dir=artifacts_dir, + status="failed", + rpms=[], + ) + raise + + print("Built RPMs:") + for rpm in all_copied_rpms: + print(rpm) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/test_server.py b/tests/test_server.py new file mode 100644 index 0000000..e04b22b --- /dev/null +++ b/tests/test_server.py @@ -0,0 +1,418 @@ +from __future__ import annotations + +import json +import os +from pathlib import Path +import tempfile +import unittest +from unittest.mock import patch + +from sailfish_devel_mcp.config import ( + BUNDLED_BUILD_HELPER, + Config, + DeviceConfig, + PathConfig, + load_config, +) +from sailfish_devel_mcp.runner import CommandResult +from sailfish_devel_mcp.server import McpServer +from sailfish_devel_mcp.tools import _device_home_path, _screenshot_prepare_command +from sailfish_devel_mcp.vendor import build_sailfishos + + +class McpServerTests(unittest.TestCase): + def make_server(self, root: Path) -> McpServer: + config = Config( + path=None, + default_device="test", + devices={"test": DeviceConfig(name="test", ssh_target="root@test")}, + paths=PathConfig( + git_root=root, + ssh_config=root / "ssh_config", + build_sailfishos=root / "tools" / "build_sailfishos.py", + osc_api_alias="jolla", + ), + ) + return McpServer(config) + + def test_initialize_uses_requested_supported_version(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + server = self.make_server(Path(tmp)) + response = server.handle( + { + "jsonrpc": "2.0", + "id": 1, + "method": "initialize", + "params": { + "protocolVersion": "2025-06-18", + "capabilities": {}, + "clientInfo": {"name": "test", "version": "0"}, + }, + } + ) + self.assertIsNotNone(response) + self.assertEqual(response["result"]["protocolVersion"], "2025-06-18") + + def test_tools_list_contains_device_and_qml_tools(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + server = self.make_server(Path(tmp)) + response = server.handle( + {"jsonrpc": "2.0", "id": 2, "method": "tools/list", "params": {}} + ) + names = {tool["name"] for tool in response["result"]["tools"]} + self.assertIn("sailfish_device_topmost_pid", names) + self.assertIn("sailfish_device_touch", names) + self.assertIn("sailfish_qml_check_translator_ternaries", names) + + def test_qml_ternary_checker_reports_inline_ternary_qstrid(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + qml = root / "Example.qml" + qml.write_text( + 'Text { text: enabled ? qsTrId("a") : qsTrId("b") }\n', + encoding="utf-8", + ) + server = self.make_server(root) + response = server.handle( + { + "jsonrpc": "2.0", + "id": 3, + "method": "tools/call", + "params": { + "name": "sailfish_qml_check_translator_ternaries", + "arguments": {"path": str(root)}, + }, + } + ) + self.assertTrue(response["result"]["isError"]) + findings = response["result"]["structuredContent"]["findings"] + self.assertEqual(len(findings), 1) + self.assertEqual(findings[0]["line"], 1) + + def test_spec_summary_parses_core_fields(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + rpm = root / "rpm" + rpm.mkdir() + spec = rpm / "sample.spec" + spec.write_text( + "\n".join( + [ + "Name: sample", + "Version: 1.2.3", + "Release: 1", + "Summary: Sample package", + "BuildRequires: pkgconfig(Qt5Core)", + "Requires: lipstick", + ] + ), + encoding="utf-8", + ) + server = self.make_server(root) + response = server.handle( + { + "jsonrpc": "2.0", + "id": 4, + "method": "tools/call", + "params": { + "name": "sailfish_spec_summary", + "arguments": {"repo_path": str(root)}, + }, + } + ) + summary = response["result"]["structuredContent"]["summary"] + self.assertEqual(summary["Name"], "sample") + self.assertEqual(summary["Version"], "1.2.3") + self.assertEqual(summary["BuildRequires"], ["pkgconfig(Qt5Core)"]) + + def test_server_writes_json_serializable_tool_result(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + server = self.make_server(Path(tmp)) + response = server.handle( + { + "jsonrpc": "2.0", + "id": 5, + "method": "tools/call", + "params": {"name": "sailfish_devices", "arguments": {}}, + } + ) + json.dumps(response) + + def test_default_config_uses_bundled_build_helper(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + missing = Path(tmp) / "missing-config.json" + with patch.dict(os.environ, {}, clear=True): + config = load_config(missing) + self.assertEqual(config.default_device, "device") + self.assertEqual(config.devices["device"].ssh_target, "root@device") + self.assertEqual(config.devices["device"].username, "defaultuser") + self.assertEqual(_device_home_path(config.devices["device"]), "/home/defaultuser") + self.assertNotIn("ssh_config", config.devices["device"].public_dict()) + self.assertEqual(config.paths.ssh_config, Path.home() / ".ssh" / "config") + self.assertEqual(config.paths.build_sailfishos, BUNDLED_BUILD_HELPER) + self.assertIsNone(config.paths.local_sdk) + self.assertTrue(config.paths.build_sailfishos.exists()) + self.assertNotIn("build-sailfishos-skill", str(config.paths.build_sailfishos)) + + def test_config_loads_local_sdk_from_paths(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + local_sdk = root / "sdk-chroot" + ssh_config = root / "ssh_config" + config_path = root / "config.json" + config_path.write_text( + json.dumps( + { + "devices": {"phone": "root@phone"}, + "paths": { + "git_root": str(root), + "ssh_config": str(ssh_config), + "local_sdk": str(local_sdk), + }, + } + ), + encoding="utf-8", + ) + config = load_config(config_path) + self.assertEqual(config.paths.local_sdk, local_sdk) + self.assertEqual(config.paths.ssh_config, ssh_config) + self.assertEqual(config.devices["phone"].release, "") + + def test_config_keeps_device_release_literal(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + config_path = root / "config.json" + config_path.write_text( + json.dumps( + { + "devices": { + "phone": { + "ssh_target": "root@phone", + "release": "devel", + } + } + } + ), + encoding="utf-8", + ) + config = load_config(config_path) + self.assertEqual(config.devices["phone"].release, "devel") + + def test_screenshot_prepare_command_uses_home_ownership(self) -> None: + device = DeviceConfig(name="test", ssh_target="root@test") + command = _screenshot_prepare_command( + device, + "/home/defaultuser/Pictures/Screenshots/example.png", + ) + self.assertIn("stat -Lc %U", command) + self.assertIn("install -d -m 775", command) + self.assertIn("install -d -m 755", command) + self.assertIn("privileged", command) + + def test_device_touch_tap_builds_input_injection_command(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + server = self.make_server(Path(tmp)) + with patch("sailfish_devel_mcp.tools.run") as mocked_run: + mocked_run.return_value = CommandResult( + ("ssh", "root@test"), + 0, + "input_device=/dev/input/event5\ntap=ok\n", + "", + ) + response = server.handle( + { + "jsonrpc": "2.0", + "id": 6, + "method": "tools/call", + "params": { + "name": "sailfish_device_touch", + "arguments": { + "action": "tap", + "input_device": "/dev/input/event5", + "x": 12, + "y": 34, + }, + }, + } + ) + self.assertFalse(response["result"].get("isError", False)) + argv = mocked_run.call_args.args[0] + remote = argv[-1] + self.assertEqual(argv[:3], ["ssh", "-F", str(Path(tmp) / "ssh_config")]) + self.assertIn("INPUT_DEVICE=/dev/input/event5", remote) + self.assertIn("X=12", remote) + self.assertIn("Y=34", remote) + self.assertIn("ABS_MT_TRACKING_ID", remote) + + def test_device_touch_discover_uses_evdev_trace_when_available(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + server = self.make_server(Path(tmp)) + with patch("sailfish_devel_mcp.tools.run") as mocked_run: + mocked_run.return_value = CommandResult(("ssh", "root@test"), 0, "", "") + response = server.handle( + { + "jsonrpc": "2.0", + "id": 7, + "method": "tools/call", + "params": { + "name": "sailfish_device_touch", + "arguments": {"action": "discover"}, + }, + } + ) + self.assertFalse(response["result"].get("isError", False)) + remote = mocked_run.call_args.args[0][-1] + self.assertNotIn("evdev_trace -i", remote) + self.assertIn("/proc/bus/input/devices", remote) + + def test_device_touch_discover_can_include_evdev_trace(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + server = self.make_server(Path(tmp)) + with patch("sailfish_devel_mcp.tools.run") as mocked_run: + mocked_run.return_value = CommandResult(("ssh", "root@test"), 0, "", "") + response = server.handle( + { + "jsonrpc": "2.0", + "id": 8, + "method": "tools/call", + "params": { + "name": "sailfish_device_touch", + "arguments": { + "action": "discover", + "include_evdev_trace": True, + }, + }, + } + ) + self.assertFalse(response["result"].get("isError", False)) + remote = mocked_run.call_args.args[0][-1] + self.assertIn("evdev_trace -i", remote) + + def test_build_defaults_prefer_configured_local_sdk(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + (root / "rpm").mkdir() + (root / "rpm" / "sample.spec").write_text("Name: sample\n", encoding="utf-8") + helper = root / "build_sailfishos.py" + local_sdk = root / "sdk-chroot" + helper.write_text("#!/usr/bin/env python3\n", encoding="utf-8") + config = Config( + path=None, + default_device="phone", + devices={ + "phone": DeviceConfig( + name="phone", + ssh_target="root@phone", + architecture="aarch64", + release="live", + ) + }, + paths=PathConfig( + git_root=root, + ssh_config=root / "ssh_config", + build_sailfishos=helper, + local_sdk=local_sdk, + ), + ) + server = McpServer(config) + with patch("sailfish_devel_mcp.tools.run") as mocked_run: + mocked_run.return_value = CommandResult(("python3", str(helper)), 0, "", "") + response = server.handle( + { + "jsonrpc": "2.0", + "id": 6, + "method": "tools/call", + "params": { + "name": "sailfish_build_rpm", + "arguments": {"project_path": str(root), "device": "phone"}, + }, + } + ) + self.assertFalse(response["result"].get("isError", False)) + argv = list(mocked_run.call_args.args[0]) + self.assertIn("--local-sdk", argv) + self.assertIn(str(local_sdk), argv) + self.assertIn("--release", argv) + self.assertIn("live", argv) + self.assertIn("--arch", argv) + self.assertIn("aarch64", argv) + + def write_fake_target( + self, + root: Path, + name: str, + *, + ssu_release: str, + version_id: str, + flavour: str = "devel", + ) -> None: + target = root / "targets" / name + (target / "etc" / "ssu").mkdir(parents=True) + (target / "etc").mkdir(exist_ok=True) + (target / "etc" / "ssu" / "ssu.ini").write_text( + f"[rnd]\nrelease={ssu_release}\nflavour={flavour}\n", + encoding="utf-8", + ) + (target / "etc" / "sailfish-release").write_text( + f'VERSION_ID={version_id}\nSAILFISH_FLAVOUR="{flavour}"\n', + encoding="utf-8", + ) + + def test_build_helper_selects_matching_local_sdk_targets(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + local_sdk = root / "sdks" / "sfossdk" / "sdk-chroot" + local_sdk.parent.mkdir(parents=True) + local_sdk.touch() + sdk_root = local_sdk.parent + self.write_fake_target(sdk_root, "aarch64", ssu_release="live", version_id="5.2.0.4") + self.write_fake_target( + sdk_root, + "aarch64-5.0.0", + ssu_release="5.0.0", + version_id="5.0.0.191", + ) + + self.assertEqual(build_sailfishos.normalize_release_tag("live"), "live") + self.assertEqual(build_sailfishos.normalize_release_tag("devel"), "devel") + self.assertEqual(build_sailfishos.normalize_local_release("devel"), "devel") + live = build_sailfishos.select_local_sdk_builds( + local_sdk, + "live", + ["aarch64"], + False, + root, + ) + versioned = build_sailfishos.select_local_sdk_builds( + local_sdk, + "5.0.0", + ["aarch64"], + False, + root, + ) + missing = build_sailfishos.select_local_sdk_builds( + local_sdk, + "4.5.0", + ["aarch64"], + False, + root, + ) + exact_mismatch = build_sailfishos.select_local_sdk_builds( + local_sdk, + "5.0.0.55", + ["aarch64"], + False, + root, + ) + + self.assertEqual(live, [build_sailfishos.LocalSdkBuild("aarch64", "aarch64")]) + self.assertEqual( + versioned, + [build_sailfishos.LocalSdkBuild("aarch64", "aarch64-5.0.0")], + ) + self.assertIsNone(missing) + self.assertIsNone(exact_mismatch) + + +if __name__ == "__main__": + unittest.main() |