summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorAndrew Branson <andrew.branson@jolla.com>2026-06-02 00:17:51 +0200
committerAndrew Branson <andrew.branson@jolla.com>2026-06-02 00:17:51 +0200
commit0cdf2d00cdc1a9af5f5b0b14241a2afe717661f6 (patch)
treea41154a24cb2c226847aef053d0121e34ff99ad6
parent1f14c5483ee111105f94d66fb3b82208946d914a (diff)
Add Sailfish MCP build and device helpers
Add OBS root scoping, user-session/browser/touch workflow device tools, local SDK metadata refresh, and asynchronous RPM build job status support.
-rw-r--r--README.md62
-rw-r--r--examples/config.json5
-rw-r--r--src/sailfish_devel_mcp/config.py9
-rw-r--r--src/sailfish_devel_mcp/tools.py738
-rwxr-xr-xsrc/sailfish_devel_mcp/vendor/build_sailfishos.py27
-rw-r--r--tests/test_server.py323
6 files changed, 1130 insertions, 34 deletions
diff --git a/README.md b/README.md
index 54e9247..23257e1 100644
--- a/README.md
+++ b/README.md
@@ -14,12 +14,16 @@ The server currently exposes tools for:
- defaultuser session-bus calls
- Lipstick screenshots
- touchscreen discovery and tap/swipe injection
+- combined screenshot, touchscreen discovery, and touch injection workflows
- topmost window PID lookup
- process map inspection
- journal log reads
- RPM copy/install on a device
- system and user service management
+- user-session command execution with the configured D-Bus environment
+- Sailfish Browser launch/debug helpers
- Docker/mb2 RPM builds through the local `build-sailfishos` helper
+- installed SDK repository metadata refresh
- Jolla OBS result and build-log lookup through `osc`
- repository status and search under the configured git root
- RPM spec metadata summaries
@@ -27,8 +31,8 @@ The server currently exposes tools for:
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
+`/run/user/100000/dbus/user_bus_socket`, `~/git` as the local source root,
+`~/OBS` as the OBS checkout 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
@@ -81,8 +85,9 @@ Example:
}
},
"paths": {
- "git_root": "/home/you/git",
- "ssh_config": "/home/you/.ssh/config",
+ "git_root": "~/git",
+ "obs_root": "~/OBS",
+ "ssh_config": "~/.ssh/config",
"local_sdk": "/srv/mer/sdks/sfossdk/sdk-chroot",
"osc_api_alias": "your-obs-alias"
}
@@ -91,18 +96,23 @@ Example:
## 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.
+The server keeps local paths scoped to the configured git root, OBS checkout
+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_touch_workflow`
- `sailfish_device_user_bus_call`
+- `sailfish_device_user_session_command`
- `sailfish_device_install_rpm`
- `sailfish_device_restart_service`
+- `sailfish_device_browser_launch`
- `sailfish_build_rpm`
+- `sailfish_sdk_refresh_metadata`
`sailfish_device_lipstick_screenshot` defaults to
`~/Pictures/Screenshots/lipstick-<timestamp>.png` under `/home/<username>`,
@@ -119,16 +129,42 @@ 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_device_touch_workflow` wraps the common UI-debugging sequence:
+capture a Lipstick screenshot, optionally list touch devices, inject a tap or
+swipe, and optionally capture a second screenshot. It returns each step's
+structured result separately.
+
+`sailfish_device_user_session_command` runs an argv command with
+`XDG_RUNTIME_DIR` and `DBUS_SESSION_BUS_ADDRESS` set from the configured device.
+Set `run_as_user` when the command should execute as the configured Sailfish
+username through `runuser` or `su`.
+
+`sailfish_device_browser_launch` stops the browser booster service and stale
+browser/firejail PIDs when requested, launches Sailfish Browser through
+`invoker` with the display and user-session environment, then queries Lipstick
+for the topmost PID and checks whether that process has `libxul.so` mapped.
+
`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
+build defaults to `live` and first checks the installed SDK targets. `live` uses
+the unversioned local target for the requested architecture, for example
+`aarch64`. A named production release such as `5.0.0` is used only when passed
+explicitly through the tool arguments, environment, or device config; it uses a
+matching versioned local target such as `aarch64-5.0.0` when available, and
+otherwise 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`.
+`sailfish_sdk_refresh_metadata` refreshes zypper metadata in the installed SDK
+main target, for example `aarch64.default`, using the same privileged Docker
+wrapper style as local SDK builds. Use it when local SDK builds fail because a
+package listed in repository metadata cannot be downloaded.
+
+`sailfish_obs_buildlog` defaults to `osc api` with `nostream=1` so a build log
+request does not become a long-running live stream. Set `nostream` to `false`
+to use `osc remotebuildlog`.
+
Read-only tools include the journal, topmost PID, process maps, OBS lookup,
repo search, spec summary, and QML checks.
diff --git a/examples/config.json b/examples/config.json
index fb473e5..96906ea 100644
--- a/examples/config.json
+++ b/examples/config.json
@@ -11,8 +11,9 @@
}
},
"paths": {
- "git_root": "/home/you/git",
- "ssh_config": "/home/you/.ssh/config",
+ "git_root": "~/git",
+ "obs_root": "~/OBS",
+ "ssh_config": "~/.ssh/config",
"local_sdk": "/srv/mer/sdks/sfossdk/sdk-chroot",
"osc_api_alias": "your-obs-alias"
}
diff --git a/src/sailfish_devel_mcp/config.py b/src/sailfish_devel_mcp/config.py
index 6515918..73740d1 100644
--- a/src/sailfish_devel_mcp/config.py
+++ b/src/sailfish_devel_mcp/config.py
@@ -39,6 +39,7 @@ class DeviceConfig:
@dataclass(frozen=True)
class PathConfig:
git_root: Path = Path.home() / "git"
+ obs_root: Path = Path.home() / "OBS"
ssh_config: Path = Path.home() / ".ssh" / "config"
build_sailfishos: Path = BUNDLED_BUILD_HELPER
local_sdk: Path | None = None
@@ -47,6 +48,7 @@ class PathConfig:
def public_dict(self) -> dict[str, str | None]:
return {
"git_root": str(self.git_root),
+ "obs_root": str(self.obs_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,
@@ -155,6 +157,13 @@ def _load_paths(raw_paths: Any) -> PathConfig:
or defaults.git_root
)
).expanduser(),
+ obs_root=Path(
+ str(
+ raw_paths.get("obs_root")
+ or os.environ.get("SAILFISH_MCP_OBS_ROOT")
+ or defaults.obs_root
+ )
+ ).expanduser(),
ssh_config=Path(
str(
raw_paths.get("ssh_config")
diff --git a/src/sailfish_devel_mcp/tools.py b/src/sailfish_devel_mcp/tools.py
index 1061e6c..854cd00 100644
--- a/src/sailfish_devel_mcp/tools.py
+++ b/src/sailfish_devel_mcp/tools.py
@@ -1,12 +1,18 @@
from __future__ import annotations
from dataclasses import dataclass
-from datetime import datetime
+from datetime import datetime, timezone
+import json
+import os
from pathlib import Path, PurePosixPath
import re
import shlex
import shutil
+import subprocess
+import sys
+import time
from typing import Any, Callable
+from urllib.parse import quote
from .config import Config, DeviceConfig
from .runner import (
@@ -19,6 +25,7 @@ from .runner import (
truncate,
user_bus_env,
)
+from .vendor import build_sailfishos
ToolHandler = Callable[[dict[str, Any]], dict[str, Any]]
@@ -45,15 +52,32 @@ def build_registry(config: Config) -> dict[str, Tool]:
lambda args: handle_device_touch(config, args),
),
Tool(
+ _spec_device_touch_workflow(),
+ lambda args: handle_device_touch_workflow(config, args),
+ ),
+ Tool(
_spec_device_user_bus_call(),
lambda args: handle_device_user_bus_call(config, args),
),
+ Tool(
+ _spec_device_user_session_command(),
+ lambda args: handle_device_user_session_command(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_device_browser_launch(),
+ lambda args: handle_device_browser_launch(config, args),
+ ),
Tool(_spec_build_rpm(), lambda args: handle_build_rpm(config, args)),
+ Tool(_spec_build_status(), lambda args: handle_build_status(config, args)),
+ Tool(
+ _spec_sdk_refresh_metadata(),
+ lambda args: handle_sdk_refresh_metadata(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)),
@@ -152,7 +176,7 @@ def handle_device_proc_maps(config: Config, args: dict[str, Any]) -> dict[str, A
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")
+ timestamp = datetime.now(timezone.utc).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"
@@ -238,6 +262,66 @@ def handle_device_touch(config: Config, args: dict[str, Any]) -> dict[str, Any]:
)
+def handle_device_touch_workflow(config: Config, args: dict[str, Any]) -> dict[str, Any]:
+ device = _device(config, args)
+ action = _enum_arg(args, "action", ["tap", "swipe"])
+ timeout = _int_arg(args, "timeout", default=30, minimum=1, maximum=300)
+ screenshot_before = _bool_arg(args, "screenshot_before", default=True)
+ screenshot_after = _bool_arg(args, "screenshot_after", default=False)
+ discover_input = _bool_arg(args, "discover_input", default=True)
+ privileged = _bool_arg(args, "privileged", default=True)
+
+ steps: list[tuple[str, dict[str, Any]]] = []
+ base_args = {"device": _optional_str(args, "device") or device.name, "timeout": timeout}
+
+ def append_step(name: str, result: dict[str, Any]) -> bool:
+ steps.append((name, result))
+ return bool(result.get("isError", False))
+
+ if screenshot_before:
+ before_args: dict[str, Any] = {
+ **base_args,
+ "privileged": privileged,
+ "remote_path": _optional_str(args, "before_remote_path")
+ or _default_screenshot_path(device, "touch-before"),
+ }
+ before_local_path = _optional_str(args, "before_local_path") or _optional_str(args, "local_path")
+ if before_local_path:
+ before_args["local_path"] = before_local_path
+ if append_step("screenshot_before", handle_device_lipstick_screenshot(config, before_args)):
+ return combined_result("touch workflow", steps)
+
+ input_device = _optional_str(args, "input_device")
+ if discover_input and not input_device:
+ discover_args = {
+ **base_args,
+ "action": "discover",
+ "include_evdev_trace": _bool_arg(args, "include_evdev_trace", default=False),
+ }
+ if append_step("touchscreen_discovery", handle_device_touch(config, discover_args)):
+ return combined_result("touch workflow", steps)
+
+ touch_args = dict(args)
+ touch_args["action"] = action
+ touch_args["timeout"] = timeout
+ if append_step("touch", handle_device_touch(config, touch_args)):
+ return combined_result("touch workflow", steps)
+
+ if screenshot_after:
+ after_args: dict[str, Any] = {
+ **base_args,
+ "privileged": privileged,
+ "remote_path": _optional_str(args, "after_remote_path")
+ or _default_screenshot_path(device, "touch-after"),
+ }
+ after_local_path = _optional_str(args, "after_local_path")
+ if after_local_path:
+ after_args["local_path"] = after_local_path
+ append_step("screenshot_after", handle_device_lipstick_screenshot(config, after_args))
+
+ return combined_result("touch workflow", steps)
+
+
def handle_device_user_bus_call(config: Config, args: dict[str, Any]) -> dict[str, Any]:
device = _device(config, args)
destination = _str_arg(args, "destination")
@@ -260,6 +344,24 @@ def handle_device_user_bus_call(config: Config, args: dict[str, Any]) -> dict[st
return command_result("user bus call", _run_ssh(config, device, command, timeout=timeout))
+def handle_device_user_session_command(config: Config, args: dict[str, Any]) -> dict[str, Any]:
+ device = _device(config, args)
+ command = _command_list_arg(args, "command")
+ timeout = _int_arg(args, "timeout", default=30, minimum=1, maximum=300)
+ run_as_user = _bool_arg(args, "run_as_user", default=False)
+ session_command = user_bus_env(device) + command
+ if run_as_user:
+ remote = _run_as_user_command(device, remote_command(session_command))
+ result = run(ssh_argv(device, config.paths.ssh_config, remote), timeout=timeout)
+ else:
+ result = _run_ssh(config, device, session_command, timeout=timeout)
+ return command_result(
+ "user session command",
+ result,
+ {"run_as_user": run_as_user, "username": device.username},
+ )
+
+
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)
@@ -304,6 +406,21 @@ def handle_device_restart_service(config: Config, args: dict[str, Any]) -> dict[
)
+def handle_device_browser_launch(config: Config, args: dict[str, Any]) -> dict[str, Any]:
+ device = _device(config, args)
+ url = _str_arg(args, "url")
+ stop_stale = _bool_arg(args, "stop_stale", default=True)
+ wait_seconds = _int_arg(args, "wait_seconds", default=3, minimum=0, maximum=60)
+ timeout = _int_arg(args, "timeout", default=60, minimum=1, maximum=300)
+ remote = _browser_launch_command(device, url, stop_stale, wait_seconds)
+ result = run(ssh_argv(device, config.paths.ssh_config, remote), timeout=timeout)
+ return command_result(
+ "browser launch",
+ result,
+ {"url": url, "stop_stale": stop_stale, "wait_seconds": wait_seconds},
+ )
+
+
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
@@ -317,6 +434,8 @@ def handle_build_rpm(config: Config, args: dict[str, Any]) -> dict[str, Any]:
artifacts_dir = _optional_str(args, "artifacts_dir")
if config.paths.local_sdk:
command += ["--local-sdk", str(config.paths.local_sdk)]
+ if not release:
+ release = "live"
if release:
command += ["--release", release]
if isinstance(arches, str):
@@ -344,7 +463,248 @@ def handle_build_rpm(config: Config, args: dict[str, Any]) -> dict[str, Any]:
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))
+ if _bool_arg(args, "wait", default=False):
+ return command_result("build Sailfish RPM", run(command, timeout=timeout))
+ job = _start_background_command("build Sailfish RPM", command, timeout=timeout)
+ text = "\n".join(
+ [
+ f"started build job {job['job_id']}",
+ f"status: {job['status_path']}",
+ f"log: {job['log_path']}",
+ "poll with sailfish_build_status",
+ ]
+ )
+ return ok_text(text, job)
+
+
+def handle_build_status(config: Config, args: dict[str, Any]) -> dict[str, Any]:
+ job_id = _optional_str(args, "job_id")
+ lines = _int_arg(args, "lines", default=80, minimum=0, maximum=1000)
+ jobs_dir = _build_jobs_dir()
+ if not job_id:
+ jobs = []
+ if jobs_dir.exists():
+ for status_path in sorted(jobs_dir.glob("*/status.json"), key=lambda p: p.stat().st_mtime):
+ status = _read_job_status(status_path)
+ if status:
+ jobs.append(status)
+ jobs = jobs[-20:]
+ text = "\n".join(
+ f"{job.get('job_id')}: {job.get('state')} {job.get('started_at', '')}"
+ for job in jobs
+ )
+ return ok_text(text or "no build jobs found", {"jobs_dir": str(jobs_dir), "jobs": jobs})
+
+ job_dir = jobs_dir / job_id
+ status_path = job_dir / "status.json"
+ if not status_path.exists():
+ return tool_error(f"unknown build job: {job_id}", {"jobs_dir": str(jobs_dir)})
+ status = _read_job_status(status_path)
+ if not status:
+ return tool_error(f"could not read build job status: {job_id}")
+ log_path = Path(str(status.get("log_path") or job_dir / "build.log"))
+ log_tail = _tail_file(log_path, lines)
+ state = str(status.get("state") or "unknown")
+ returncode = status.get("returncode")
+ text_lines = [
+ f"job_id: {status.get('job_id')}",
+ f"state: {state}",
+ f"returncode: {returncode}",
+ f"log: {log_path}",
+ ]
+ if log_tail:
+ text_lines += ["", log_tail]
+ structured = dict(status)
+ structured["log_tail"] = log_tail
+ return {
+ "content": [{"type": "text", "text": "\n".join(text_lines)}],
+ "structuredContent": structured,
+ "isError": state == "finished" and returncode not in (0, None),
+ }
+
+
+def _mcp_state_dir() -> Path:
+ value = os.environ.get("SAILFISH_DEVEL_MCP_STATE_DIR")
+ if value:
+ return Path(value).expanduser()
+ value = os.environ.get("SAILFISH_DEVEL_MCP_LOG_DIR")
+ if value:
+ return Path(value).expanduser()
+ value = os.environ.get("XDG_STATE_HOME")
+ if value:
+ return Path(value).expanduser() / "sailfish-devel-mcp"
+ value = os.environ.get("HOME")
+ if value:
+ return Path(value).expanduser() / ".local" / "state" / "sailfish-devel-mcp"
+ return Path("/tmp") / "sailfish-devel-mcp"
+
+
+def _build_jobs_dir() -> Path:
+ return _mcp_state_dir() / "builds"
+
+
+def _write_json_atomic(path: Path, data: dict[str, Any]) -> None:
+ path.parent.mkdir(parents=True, exist_ok=True)
+ tmp = path.with_suffix(path.suffix + ".tmp")
+ tmp.write_text(json.dumps(data, indent=2, sort_keys=True) + "\n", encoding="utf-8")
+ tmp.replace(path)
+
+
+def _start_background_command(label: str, command: list[str], *, timeout: int) -> dict[str, Any]:
+ now = datetime.now(timezone.utc)
+ job_id = f"build-{now.strftime('%Y%m%dT%H%M%S')}-{os.getpid()}-{int(time.time() * 1000) % 100000}"
+ job_dir = _build_jobs_dir() / job_id
+ log_path = job_dir / "build.log"
+ status_path = job_dir / "status.json"
+ status = {
+ "job_id": job_id,
+ "label": label,
+ "state": "starting",
+ "argv": command,
+ "timeout": timeout,
+ "created_at": now.isoformat(),
+ "status_path": str(status_path),
+ "log_path": str(log_path),
+ }
+ _write_json_atomic(status_path, status)
+
+ supervisor = r"""
+from __future__ import annotations
+
+from datetime import datetime, timezone
+import json
+import os
+from pathlib import Path
+import signal
+import subprocess
+import sys
+import time
+
+
+def now() -> str:
+ return datetime.now(timezone.utc).isoformat()
+
+
+def write_status(path: Path, data: dict[str, object]) -> None:
+ tmp = path.with_suffix(path.suffix + ".tmp")
+ tmp.write_text(json.dumps(data, indent=2, sort_keys=True) + "\n", encoding="utf-8")
+ tmp.replace(path)
+
+
+status_path = Path(sys.argv[1])
+log_path = Path(sys.argv[2])
+timeout = int(sys.argv[3])
+command = sys.argv[4:]
+
+status = json.loads(status_path.read_text(encoding="utf-8"))
+status["supervisor_pid"] = os.getpid()
+status["started_at"] = now()
+log_path.parent.mkdir(parents=True, exist_ok=True)
+
+with log_path.open("a", encoding="utf-8", errors="replace") as log:
+ log.write(f"[{now()}] starting {' '.join(command)}\n")
+ log.flush()
+ process = subprocess.Popen(
+ command,
+ stdin=subprocess.DEVNULL,
+ stdout=log,
+ stderr=subprocess.STDOUT,
+ text=True,
+ start_new_session=True,
+ close_fds=True,
+ )
+ status["pid"] = process.pid
+ status["state"] = "running"
+ write_status(status_path, status)
+
+ deadline = time.monotonic() + timeout
+ timed_out = False
+ returncode = None
+ while True:
+ returncode = process.poll()
+ if returncode is not None:
+ break
+ if time.monotonic() >= deadline:
+ timed_out = True
+ log.write(f"[{now()}] timeout after {timeout}s; terminating process group {process.pid}\n")
+ log.flush()
+ try:
+ os.killpg(process.pid, signal.SIGTERM)
+ except ProcessLookupError:
+ pass
+ try:
+ returncode = process.wait(timeout=30)
+ except subprocess.TimeoutExpired:
+ log.write(f"[{now()}] process group did not exit; killing {process.pid}\n")
+ log.flush()
+ try:
+ os.killpg(process.pid, signal.SIGKILL)
+ except ProcessLookupError:
+ pass
+ returncode = process.wait()
+ break
+ time.sleep(1)
+
+ status["state"] = "finished"
+ status["returncode"] = returncode
+ status["timed_out"] = timed_out
+ status["finished_at"] = now()
+ write_status(status_path, status)
+ log.write(f"[{now()}] finished returncode={returncode} timed_out={timed_out}\n")
+"""
+ process = subprocess.Popen(
+ [sys.executable, "-c", supervisor, str(status_path), str(log_path), str(timeout), *command],
+ stdin=subprocess.DEVNULL,
+ stdout=subprocess.DEVNULL,
+ stderr=subprocess.DEVNULL,
+ start_new_session=True,
+ close_fds=True,
+ )
+ status["supervisor_pid"] = process.pid
+ _write_json_atomic(status_path, status)
+ return status
+
+
+def _read_job_status(path: Path) -> dict[str, Any] | None:
+ try:
+ data = json.loads(path.read_text(encoding="utf-8"))
+ except (OSError, json.JSONDecodeError):
+ return None
+ return data if isinstance(data, dict) else None
+
+
+def _tail_file(path: Path, lines: int) -> str:
+ if lines <= 0:
+ return ""
+ try:
+ text = path.read_text(encoding="utf-8", errors="replace")
+ except OSError:
+ return ""
+ return "\n".join(text.splitlines()[-lines:])
+
+
+def handle_sdk_refresh_metadata(config: Config, args: dict[str, Any]) -> dict[str, Any]:
+ local_sdk = _local_sdk_path(config, args)
+ if local_sdk is None:
+ return tool_error("paths.local_sdk or local_sdk argument is required")
+ target = _local_sdk_target(config, args, local_sdk)
+ if target is None:
+ return tool_error("target, arch, or a device with architecture is required")
+ main_target = _main_sdk_target(target)
+ timeout = _int_arg(args, "timeout", default=600, minimum=1, maximum=3600)
+ command = ["sb2", "-t", main_target, "-m", "sdk-install", "-R", "zypper", "ref"]
+ if _bool_arg(args, "force", default=False):
+ command.append("-f")
+ result = run(_local_sdk_docker_argv(local_sdk, command), timeout=timeout)
+ return command_result(
+ "refresh local SDK metadata",
+ result,
+ {
+ "local_sdk": str(local_sdk),
+ "target": target,
+ "main_target": main_target,
+ },
+ )
def handle_obs_results(config: Config, args: dict[str, Any]) -> dict[str, Any]:
@@ -368,10 +728,20 @@ def handle_obs_buildlog(config: Config, args: dict[str, Any]) -> dict[str, Any]:
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)
+ nostream = _bool_arg(args, "nostream", default=True)
command = ["osc"]
if api_alias:
command += ["-A", api_alias]
- command += ["remotebuildlog", project, package, repository, arch]
+ if nostream:
+ path = "/build/{}/{}/{}/{}/_log?nostream=1".format(
+ quote(project, safe=""),
+ quote(repository, safe=""),
+ quote(arch, safe=""),
+ quote(package, safe=""),
+ )
+ command += ["api", path]
+ else:
+ command += ["remotebuildlog", project, package, repository, arch]
return command_result("OBS build log", run(command, timeout=timeout))
@@ -489,6 +859,11 @@ def _run_ssh(
)
+def _default_screenshot_path(device: DeviceConfig, label: str) -> str:
+ timestamp = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S-%f")
+ return f"{_device_home_path(device)}/Pictures/Screenshots/{label}-{timestamp}.png"
+
+
def _screenshot_prepare_command(device: DeviceConfig, remote_path: str) -> str:
remote_dir = str(PurePosixPath(remote_path).parent)
home_path = _device_home_path(device)
@@ -519,6 +894,160 @@ fi
return remote_command(["sh", "-lc", script])
+def _run_as_user_command(device: DeviceConfig, inner: str) -> str:
+ if not re.fullmatch(r"[A-Za-z0-9_.-]+", device.username):
+ raise ValueError("device username contains unsupported characters")
+ username = shlex.quote(device.username)
+ quoted_inner = shlex.quote(inner)
+ script = f"""
+if command -v runuser >/dev/null 2>&1; then
+ runuser -u {username} -- {inner}
+else
+ su -s /bin/sh {username} -c {quoted_inner}
+fi
+""".strip()
+ return remote_command(["sh", "-lc", script])
+
+
+def _browser_launch_command(
+ device: DeviceConfig,
+ url: str,
+ stop_stale: bool,
+ wait_seconds: int,
+) -> str:
+ script = f"""
+set -u
+export XDG_RUNTIME_DIR={shlex.quote(device.user_bus_runtime_dir)}
+export DBUS_SESSION_BUS_ADDRESS={shlex.quote(device.user_bus_address)}
+if [ {shlex.quote("1" if stop_stale else "0")} = 1 ]; then
+ systemctl --user stop booster-browser@sailfish-browser.service >/dev/null 2>&1 || true
+ browser_name=sailfish-browser
+ for pid in $(pidof "$browser_name" 2>/dev/null || true); do
+ kill "$pid" >/dev/null 2>&1 || true
+ done
+ if command -v ps >/dev/null 2>&1; then
+ firejail_name=firejail
+ ps -eo pid=,comm=,args= 2>/dev/null |
+ awk -v c="$firejail_name" -v b="$browser_name" '$2 == c && index($0, b) {{ print $1 }}' |
+ while read -r pid; do
+ [ -n "$pid" ] || continue
+ kill "$pid" >/dev/null 2>&1 || true
+ done
+ fi
+fi
+WAYLAND_DISPLAY=../../display/wayland-0 QT_QPA_PLATFORM=hwcomposer \\
+ /usr/bin/invoker --type=browser,silica-qt5 -A -- \\
+ /usr/bin/sailfish-browser {shlex.quote(url)} \\
+ >/tmp/sailfish-browser-mcp.log 2>&1 &
+launcher_pid=$!
+echo "launcher_pid=$launcher_pid"
+echo "launch_log=/tmp/sailfish-browser-mcp.log"
+if [ {wait_seconds} -gt 0 ]; then
+ sleep {wait_seconds}
+fi
+reply=$(dbus-send --session --print-reply \\
+ --dest=org.nemomobile.lipstick \\
+ / \\
+ org.freedesktop.DBus.Properties.Get \\
+ string:org.nemomobile.compositor \\
+ string:privateTopmostWindowProcessId 2>/dev/null || true)
+topmost_pid=$(printf '%s\\n' "$reply" | awk '/\\b(int32|uint32|int64|uint64)\\b/ {{ print $2; exit }}')
+if [ -n "$topmost_pid" ]; then
+ echo "topmost_pid=$topmost_pid"
+ if [ -r "/proc/$topmost_pid/maps" ]; then
+ if grep -q 'libxul\\.so' "/proc/$topmost_pid/maps"; then
+ echo "libxul=present"
+ else
+ echo "libxul=absent"
+ fi
+ fi
+else
+ echo "topmost_pid="
+fi
+""".strip()
+ return remote_command(["sh", "-lc", script])
+
+
+def _local_sdk_path(config: Config, args: dict[str, Any]) -> Path | None:
+ value = _optional_str(args, "local_sdk")
+ if value:
+ return Path(value).expanduser().resolve(strict=False)
+ if config.paths.local_sdk:
+ return config.paths.local_sdk.expanduser().resolve(strict=False)
+ return None
+
+
+def _local_sdk_target(config: Config, args: dict[str, Any], local_sdk: Path) -> str | None:
+ explicit_target = _optional_str(args, "target")
+ if explicit_target:
+ return build_sailfishos.canonical_local_target_name(explicit_target)
+
+ device = config.device(_optional_str(args, "device")) if args.get("device") else None
+ arch = _optional_str(args, "arch") or (device.architecture if device else "")
+ if not arch:
+ return None
+ release = _optional_str(args, "release") or (device.release if device else "") or "live"
+ normalized_release = build_sailfishos.normalize_local_release(release) or "live"
+ matching = [
+ target
+ for target in build_sailfishos.list_local_sdk_targets(local_sdk)
+ if target.arch == arch
+ and build_sailfishos.local_target_matches_release(target, normalized_release)
+ ]
+ if matching:
+ return matching[0].target
+ if normalized_release == "live":
+ return arch
+ return f"{arch}-{normalized_release}"
+
+
+def _main_sdk_target(target: str) -> str:
+ if target.endswith(".default"):
+ return target
+ for suffix in (".build-sailfishos-skill", ".build"):
+ if target.endswith(suffix):
+ target = target[: -len(suffix)]
+ break
+ return f"{target}.default"
+
+
+def _local_sdk_docker_argv(local_sdk: Path, command: list[str]) -> list[str]:
+ user = build_sailfishos.host_user()
+ uid = os.getuid()
+ gid = os.getgid()
+ home = str(Path.home().resolve())
+ image = build_sailfishos.local_sdk_build_engine_image(user)
+ sdk_mount_root = build_sailfishos.local_sdk_mount_root(local_sdk)
+ inner = remote_command(command)
+ wrapper_command = f"""
+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)} {inner}
+""".strip()
+ return [
+ "docker",
+ "run",
+ "--rm",
+ "--privileged",
+ "-v",
+ f"{sdk_mount_root}:{sdk_mount_root}",
+ "-e",
+ f"LOCAL_SDK={local_sdk}",
+ image,
+ "bash",
+ "-lc",
+ wrapper_command,
+ ]
+
+
def _touch_discover_command(include_evdev_trace: bool) -> str:
evdev_trace = """
if command -v evdev_trace >/dev/null 2>&1; then
@@ -758,6 +1287,27 @@ def command_result(
}
+def combined_result(title: str, steps: list[tuple[str, dict[str, Any]]]) -> dict[str, Any]:
+ text_parts: list[str] = []
+ structured_steps: dict[str, Any] = {}
+ is_error = False
+ for name, result in steps:
+ content = result.get("content") or []
+ step_text = content[0].get("text", "") if content else ""
+ text_parts.append(f"{name}:\n{step_text}".rstrip())
+ structured_steps[name] = result.get("structuredContent", {})
+ is_error = is_error or bool(result.get("isError", False))
+ return {
+ "content": [{"type": "text", "text": "\n\n".join(text_parts)}],
+ "structuredContent": {
+ "title": title,
+ "step_order": [name for name, _ in steps],
+ "steps": structured_steps,
+ },
+ "isError": is_error,
+ }
+
+
def ok_text(text: str, structured: dict[str, Any] | None = None) -> dict[str, Any]:
return {
"content": [{"type": "text", "text": text}],
@@ -850,6 +1400,17 @@ def _string_list_arg(args: dict[str, Any], name: str) -> list[str]:
return value
+def _command_list_arg(args: dict[str, Any], name: str) -> list[str]:
+ value = args.get(name)
+ if (
+ not isinstance(value, list)
+ or not value
+ or not all(isinstance(item, str) and item for item in value)
+ ):
+ raise ValueError(f"{name} must be a non-empty 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):
@@ -862,7 +1423,7 @@ def _safe_input_path(config: Config, value: str, *, allow_tmp: bool) -> Path:
if not path.is_absolute():
path = config.paths.git_root / path
path = path.resolve(strict=False)
- roots = [config.paths.git_root.resolve(strict=False)]
+ roots = _host_path_roots(config)
if allow_tmp:
roots.append(Path("/tmp").resolve(strict=False))
if not _is_relative_to_any(path, roots):
@@ -875,12 +1436,21 @@ def _safe_output_path(config: Config, value: str) -> Path:
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)]
+ roots = _host_path_roots(config) + [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 _host_path_roots(config: Config) -> list[Path]:
+ roots: list[Path] = []
+ for root in (config.paths.git_root, config.paths.obs_root):
+ resolved = root.resolve(strict=False)
+ if resolved not in roots:
+ roots.append(resolved)
+ return roots
+
+
def _is_relative_to_any(path: Path, roots: list[Path]) -> bool:
for root in roots:
try:
@@ -1083,6 +1653,49 @@ def _spec_device_touch() -> dict[str, Any]:
}
+def _spec_device_touch_workflow() -> dict[str, Any]:
+ return {
+ "name": "sailfish_device_touch_workflow",
+ "title": "Screenshot And Touch",
+ "description": "Capture a Lipstick screenshot, optionally list touch inputs, then inject a tap or swipe.",
+ "inputSchema": _object_schema(
+ {
+ "device": _device_prop(),
+ "action": {"type": "string", "enum": ["tap", "swipe"]},
+ "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},
+ "screenshot_before": {"type": "boolean", "default": True},
+ "screenshot_after": {"type": "boolean", "default": False},
+ "discover_input": {"type": "boolean", "default": True},
+ "include_evdev_trace": {"type": "boolean", "default": False},
+ "privileged": {"type": "boolean", "default": True},
+ "local_path": {
+ "type": "string",
+ "description": "Optional local path for the before screenshot.",
+ },
+ "before_local_path": {"type": "string"},
+ "after_local_path": {"type": "string"},
+ "before_remote_path": {"type": "string"},
+ "after_remote_path": {"type": "string"},
+ "timeout": _timeout_prop(30),
+ },
+ ["action"],
+ ),
+ "annotations": _mutating_annotations("Screenshot And Touch"),
+ }
+
+
def _spec_device_user_bus_call() -> dict[str, Any]:
return {
"name": "sailfish_device_user_bus_call",
@@ -1108,6 +1721,33 @@ def _spec_device_user_bus_call() -> dict[str, Any]:
}
+def _spec_device_user_session_command() -> dict[str, Any]:
+ return {
+ "name": "sailfish_device_user_session_command",
+ "title": "User Session Command",
+ "description": "Run a command with the configured Sailfish user-session D-Bus environment.",
+ "inputSchema": _object_schema(
+ {
+ "device": _device_prop(),
+ "command": {
+ "type": "array",
+ "items": {"type": "string"},
+ "minItems": 1,
+ "description": "Command argv to run; shell parsing is not applied.",
+ },
+ "run_as_user": {
+ "type": "boolean",
+ "default": False,
+ "description": "Run through runuser/su as the configured device username.",
+ },
+ "timeout": _timeout_prop(30),
+ },
+ ["command"],
+ ),
+ "annotations": _mutating_annotations("User Session Command"),
+ }
+
+
def _spec_device_install_rpm() -> dict[str, Any]:
return {
"name": "sailfish_device_install_rpm",
@@ -1150,11 +1790,30 @@ def _spec_device_restart_service() -> dict[str, Any]:
}
+def _spec_device_browser_launch() -> dict[str, Any]:
+ return {
+ "name": "sailfish_device_browser_launch",
+ "title": "Launch Sailfish Browser",
+ "description": "Stop stale browser state, launch Sailfish Browser with display/session env, and report topmost PID details.",
+ "inputSchema": _object_schema(
+ {
+ "device": _device_prop(),
+ "url": {"type": "string"},
+ "stop_stale": {"type": "boolean", "default": True},
+ "wait_seconds": {"type": "integer", "minimum": 0, "maximum": 60, "default": 3},
+ "timeout": _timeout_prop(60),
+ },
+ ["url"],
+ ),
+ "annotations": _mutating_annotations("Launch Sailfish Browser"),
+ }
+
+
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.",
+ "description": "Start an asynchronous RPM build; paths.local_sdk defaults to the live installed SDK, with public SDK fallback only for explicit named releases.",
"inputSchema": _object_schema(
{
"project_path": {"type": "string"},
@@ -1175,6 +1834,11 @@ def _spec_build_rpm() -> dict[str, Any]:
"debug": {"type": "boolean", "default": False},
"no_pull": {"type": "boolean", "default": False},
"local_rpms_dir": {"type": "array", "items": {"type": "string"}},
+ "wait": {
+ "type": "boolean",
+ "default": False,
+ "description": "Wait for the build and return the old synchronous command result.",
+ },
"timeout": _timeout_prop(3600),
},
["project_path"],
@@ -1183,6 +1847,63 @@ def _spec_build_rpm() -> dict[str, Any]:
}
+def _spec_build_status() -> dict[str, Any]:
+ return {
+ "name": "sailfish_build_status",
+ "title": "Build Job Status",
+ "description": "Read status and recent log output for asynchronous Sailfish RPM build jobs.",
+ "inputSchema": _object_schema(
+ {
+ "job_id": {
+ "type": "string",
+ "description": "Build job id returned by sailfish_build_rpm. Omit to list recent jobs.",
+ },
+ "lines": {
+ "type": "integer",
+ "minimum": 0,
+ "maximum": 1000,
+ "default": 80,
+ "description": "Number of trailing log lines to include.",
+ },
+ }
+ ),
+ "annotations": _read_only_annotations("Build Job Status"),
+ }
+
+
+def _spec_sdk_refresh_metadata() -> dict[str, Any]:
+ return {
+ "name": "sailfish_sdk_refresh_metadata",
+ "title": "Refresh SDK Metadata",
+ "description": "Refresh zypper repository metadata in an installed local SDK main target.",
+ "inputSchema": _object_schema(
+ {
+ "device": {
+ "type": "string",
+ "description": "Optional configured device to supply default release and architecture.",
+ },
+ "local_sdk": {
+ "type": "string",
+ "description": "Override paths.local_sdk for this call.",
+ },
+ "target": {
+ "type": "string",
+ "description": "Local SDK target base or .default target, for example aarch64 or aarch64.default.",
+ },
+ "release": {"type": "string"},
+ "arch": {"type": "string"},
+ "force": {
+ "type": "boolean",
+ "default": False,
+ "description": "Append zypper ref -f.",
+ },
+ "timeout": _timeout_prop(600),
+ }
+ ),
+ "annotations": _mutating_annotations("Refresh SDK Metadata"),
+ }
+
+
def _spec_obs_results() -> dict[str, Any]:
return {
"name": "sailfish_obs_results",
@@ -1205,7 +1926,7 @@ 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.",
+ "description": "Fetch an OBS build log with osc; defaults to the API nostream form to avoid live streams.",
"inputSchema": _object_schema(
{
"project": {"type": "string"},
@@ -1213,6 +1934,7 @@ def _spec_obs_buildlog() -> dict[str, Any]:
"repository": {"type": "string"},
"arch": {"type": "string"},
"api_alias": {"type": "string"},
+ "nostream": {"type": "boolean", "default": True},
"timeout": _timeout_prop(90),
},
["project", "package", "repository", "arch"],
diff --git a/src/sailfish_devel_mcp/vendor/build_sailfishos.py b/src/sailfish_devel_mcp/vendor/build_sailfishos.py
index 98b52c0..c9ad480 100755
--- a/src/sailfish_devel_mcp/vendor/build_sailfishos.py
+++ b/src/sailfish_devel_mcp/vendor/build_sailfishos.py
@@ -848,12 +848,9 @@ def normalize_local_release(release: str | None) -> str:
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
+ requested = explicit_requested_release(explicit_release)
+ if requested:
+ return requested
seen: set[Path] = set()
for project_dir in project_dirs:
@@ -867,6 +864,21 @@ def requested_release(project_dirs: Iterable[Path], explicit_release: str | None
return None
+def explicit_requested_release(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
+
+ return None
+
+
+def local_sdk_requested_release(explicit_release: str | None) -> str:
+ return normalize_local_release(explicit_requested_release(explicit_release)) or LIVE_RELEASE
+
+
def select_local_sdk_builds(
local_sdk: Path,
release: str,
@@ -1334,11 +1346,10 @@ def main() -> int:
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_release = local_sdk_requested_release(args.release)
local_builds = select_local_sdk_builds(
local_sdk_path,
local_release,
diff --git a/tests/test_server.py b/tests/test_server.py
index e04b22b..26d01b7 100644
--- a/tests/test_server.py
+++ b/tests/test_server.py
@@ -28,6 +28,7 @@ class McpServerTests(unittest.TestCase):
devices={"test": DeviceConfig(name="test", ssh_target="root@test")},
paths=PathConfig(
git_root=root,
+ obs_root=root / "OBS",
ssh_config=root / "ssh_config",
build_sailfishos=root / "tools" / "build_sailfishos.py",
osc_api_alias="jolla",
@@ -62,6 +63,10 @@ class McpServerTests(unittest.TestCase):
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_device_touch_workflow", names)
+ self.assertIn("sailfish_device_user_session_command", names)
+ self.assertIn("sailfish_device_browser_launch", names)
+ self.assertIn("sailfish_sdk_refresh_metadata", names)
self.assertIn("sailfish_qml_check_translator_ternaries", names)
def test_qml_ternary_checker_reports_inline_ternary_qstrid(self) -> None:
@@ -149,6 +154,7 @@ class McpServerTests(unittest.TestCase):
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.obs_root, Path.home() / "OBS")
self.assertEqual(config.paths.build_sailfishos, BUNDLED_BUILD_HELPER)
self.assertIsNone(config.paths.local_sdk)
self.assertTrue(config.paths.build_sailfishos.exists())
@@ -158,6 +164,7 @@ class McpServerTests(unittest.TestCase):
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
local_sdk = root / "sdk-chroot"
+ obs_root = root / "OBS"
ssh_config = root / "ssh_config"
config_path = root / "config.json"
config_path.write_text(
@@ -166,6 +173,7 @@ class McpServerTests(unittest.TestCase):
"devices": {"phone": "root@phone"},
"paths": {
"git_root": str(root),
+ "obs_root": str(obs_root),
"ssh_config": str(ssh_config),
"local_sdk": str(local_sdk),
},
@@ -175,6 +183,7 @@ class McpServerTests(unittest.TestCase):
)
config = load_config(config_path)
self.assertEqual(config.paths.local_sdk, local_sdk)
+ self.assertEqual(config.paths.obs_root, obs_root)
self.assertEqual(config.paths.ssh_config, ssh_config)
self.assertEqual(config.devices["phone"].release, "")
@@ -198,6 +207,35 @@ class McpServerTests(unittest.TestCase):
config = load_config(config_path)
self.assertEqual(config.devices["phone"].release, "devel")
+ def test_repo_find_allows_obs_root_paths(self) -> None:
+ with tempfile.TemporaryDirectory() as tmp:
+ root = Path(tmp)
+ obs_checkout = root / "OBS" / "home:example" / "browser"
+ obs_checkout.mkdir(parents=True)
+ server = self.make_server(root)
+ with (
+ patch("sailfish_devel_mcp.tools.shutil.which", return_value="/usr/bin/rg"),
+ patch("sailfish_devel_mcp.tools.run") as mocked_run,
+ ):
+ mocked_run.return_value = CommandResult(("rg",), 0, "match\n", "")
+ response = server.handle(
+ {
+ "jsonrpc": "2.0",
+ "id": 6,
+ "method": "tools/call",
+ "params": {
+ "name": "sailfish_repo_find",
+ "arguments": {
+ "path": str(obs_checkout),
+ "query": "Package",
+ },
+ },
+ }
+ )
+ self.assertFalse(response["result"].get("isError", False))
+ argv = list(mocked_run.call_args.args[0])
+ self.assertEqual(argv[-1], str(obs_checkout))
+
def test_screenshot_prepare_command_uses_home_ownership(self) -> None:
device = DeviceConfig(name="test", ssh_target="root@test")
command = _screenshot_prepare_command(
@@ -288,6 +326,169 @@ class McpServerTests(unittest.TestCase):
remote = mocked_run.call_args.args[0][-1]
self.assertIn("evdev_trace -i", remote)
+ def test_device_touch_workflow_screenshots_discovers_and_taps(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.side_effect = [
+ CommandResult(("ssh", "root@test"), 0, "screenshot\n", ""),
+ CommandResult(("ssh", "root@test"), 0, "inputs\n", ""),
+ CommandResult(("ssh", "root@test"), 0, "tap=ok\n", ""),
+ ]
+ response = server.handle(
+ {
+ "jsonrpc": "2.0",
+ "id": 9,
+ "method": "tools/call",
+ "params": {
+ "name": "sailfish_device_touch_workflow",
+ "arguments": {"action": "tap", "x": 12, "y": 34},
+ },
+ }
+ )
+ self.assertFalse(response["result"].get("isError", False))
+ self.assertEqual(
+ response["result"]["structuredContent"]["step_order"],
+ ["screenshot_before", "touchscreen_discovery", "touch"],
+ )
+ remotes = [call.args[0][-1] for call in mocked_run.call_args_list]
+ self.assertIn("saveScreenshot", remotes[0])
+ self.assertIn("/proc/bus/input/devices", remotes[1])
+ self.assertIn("X=12", remotes[2])
+ self.assertIn("Y=34", remotes[2])
+
+ def test_user_session_command_can_run_as_defaultuser(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": 10,
+ "method": "tools/call",
+ "params": {
+ "name": "sailfish_device_user_session_command",
+ "arguments": {
+ "command": ["systemctl", "--user", "status", "app.service"],
+ "run_as_user": True,
+ },
+ },
+ }
+ )
+ self.assertFalse(response["result"].get("isError", False))
+ remote = mocked_run.call_args.args[0][-1]
+ self.assertIn("runuser -u defaultuser -- env", remote)
+ self.assertIn("XDG_RUNTIME_DIR=/run/user/100000", remote)
+ self.assertIn("DBUS_SESSION_BUS_ADDRESS=unix:path=/run/user/100000/dbus/user_bus_socket", remote)
+ self.assertIn("systemctl --user status app.service", remote)
+
+ def test_browser_launch_uses_display_env_and_reports_topmost_pid(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": 11,
+ "method": "tools/call",
+ "params": {
+ "name": "sailfish_device_browser_launch",
+ "arguments": {"url": "https://example.com"},
+ },
+ }
+ )
+ self.assertFalse(response["result"].get("isError", False))
+ remote = mocked_run.call_args.args[0][-1]
+ self.assertIn("WAYLAND_DISPLAY=../../display/wayland-0", remote)
+ self.assertIn("/usr/bin/invoker --type=browser,silica-qt5 -A --", remote)
+ self.assertIn("/usr/bin/sailfish-browser https://example.com", remote)
+ self.assertIn("privateTopmostWindowProcessId", remote)
+
+ def test_obs_buildlog_uses_nostream_api_by_default(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(("osc",), 0, "", "")
+ response = server.handle(
+ {
+ "jsonrpc": "2.0",
+ "id": 12,
+ "method": "tools/call",
+ "params": {
+ "name": "sailfish_obs_buildlog",
+ "arguments": {
+ "project": "home:example",
+ "package": "browser",
+ "repository": "5.0.0",
+ "arch": "aarch64",
+ },
+ },
+ }
+ )
+ self.assertFalse(response["result"].get("isError", False))
+ argv = list(mocked_run.call_args.args[0])
+ self.assertEqual(argv[:3], ["osc", "-A", "jolla"])
+ self.assertEqual(argv[3], "api")
+ self.assertEqual(
+ argv[4],
+ "/build/home%3Aexample/5.0.0/aarch64/browser/_log?nostream=1",
+ )
+
+ def test_sdk_refresh_metadata_uses_local_sdk_main_target(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()
+ self.write_fake_target(
+ local_sdk.parent,
+ "aarch64",
+ ssu_release="live",
+ version_id="5.2.0.4",
+ )
+ 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",
+ local_sdk=local_sdk,
+ ),
+ )
+ server = McpServer(config)
+ with patch("sailfish_devel_mcp.tools.run") as mocked_run:
+ mocked_run.return_value = CommandResult(("docker",), 0, "", "")
+ response = server.handle(
+ {
+ "jsonrpc": "2.0",
+ "id": 13,
+ "method": "tools/call",
+ "params": {
+ "name": "sailfish_sdk_refresh_metadata",
+ "arguments": {"device": "phone"},
+ },
+ }
+ )
+ self.assertFalse(response["result"].get("isError", False))
+ argv = list(mocked_run.call_args.args[0])
+ self.assertEqual(argv[:3], ["docker", "run", "--rm"])
+ wrapper = argv[-1]
+ self.assertIn("sb2 -t aarch64.default -m sdk-install -R zypper ref", wrapper)
+ self.assertEqual(
+ response["result"]["structuredContent"]["main_target"],
+ "aarch64.default",
+ )
+
def test_build_defaults_prefer_configured_local_sdk(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
@@ -315,8 +516,12 @@ class McpServerTests(unittest.TestCase):
),
)
server = McpServer(config)
- with patch("sailfish_devel_mcp.tools.run") as mocked_run:
- mocked_run.return_value = CommandResult(("python3", str(helper)), 0, "", "")
+ with patch("sailfish_devel_mcp.tools._start_background_command") as mocked_start:
+ mocked_start.return_value = {
+ "job_id": "build-test",
+ "status_path": str(root / "status.json"),
+ "log_path": str(root / "build.log"),
+ }
response = server.handle(
{
"jsonrpc": "2.0",
@@ -329,13 +534,108 @@ class McpServerTests(unittest.TestCase):
}
)
self.assertFalse(response["result"].get("isError", False))
- argv = list(mocked_run.call_args.args[0])
+ argv = list(mocked_start.call_args.args[1])
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)
+ self.assertEqual(
+ response["result"]["structuredContent"]["job_id"],
+ "build-test",
+ )
+
+ def test_build_without_release_defaults_to_live_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")},
+ 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._start_background_command") as mocked_start:
+ mocked_start.return_value = {
+ "job_id": "build-test",
+ "status_path": str(root / "status.json"),
+ "log_path": str(root / "build.log"),
+ }
+ response = server.handle(
+ {
+ "jsonrpc": "2.0",
+ "id": 14,
+ "method": "tools/call",
+ "params": {
+ "name": "sailfish_build_rpm",
+ "arguments": {
+ "project_path": str(root),
+ "arch": "aarch64",
+ },
+ },
+ }
+ )
+ self.assertFalse(response["result"].get("isError", False))
+ argv = list(mocked_start.call_args.args[1])
+ 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 test_build_wait_true_uses_synchronous_runner(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"
+ 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")},
+ paths=PathConfig(
+ git_root=root,
+ ssh_config=root / "ssh_config",
+ build_sailfishos=helper,
+ ),
+ )
+ server = McpServer(config)
+ with patch("sailfish_devel_mcp.tools.run") as mocked_run:
+ mocked_run.return_value = CommandResult(("python3", str(helper)), 0, "ok", "")
+ response = server.handle(
+ {
+ "jsonrpc": "2.0",
+ "id": 15,
+ "method": "tools/call",
+ "params": {
+ "name": "sailfish_build_rpm",
+ "arguments": {
+ "project_path": str(root),
+ "arch": "aarch64",
+ "wait": True,
+ },
+ },
+ }
+ )
+
+ self.assertFalse(response["result"].get("isError", False))
+ self.assertEqual(response["result"]["structuredContent"]["returncode"], 0)
+ argv = list(mocked_run.call_args.args[0])
+ self.assertIn("--arch", argv)
+ self.assertIn("aarch64", argv)
def write_fake_target(
self,
@@ -413,6 +713,23 @@ class McpServerTests(unittest.TestCase):
self.assertIsNone(missing)
self.assertIsNone(exact_mismatch)
+ def test_build_helper_local_sdk_default_ignores_workflow_release(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")
+ workflows = root / ".github" / "workflows"
+ workflows.mkdir(parents=True)
+ (workflows / "build.yml").write_text("env:\n RELEASE: 5.0.0\n", encoding="utf-8")
+
+ with patch.dict(os.environ, {}, clear=True):
+ self.assertEqual(
+ build_sailfishos.requested_release((root,), None),
+ "5.0.0",
+ )
+ self.assertIsNone(build_sailfishos.explicit_requested_release(None))
+ self.assertEqual(build_sailfishos.local_sdk_requested_release(None), "live")
+
if __name__ == "__main__":
unittest.main()