Compare commits
3 Commits
032bc3a580
..
main
| Author | SHA1 | Date | |
|---|---|---|---|
| 1996a123d9 | |||
| c445093186 | |||
| bdf3b55a98 |
@@ -4,16 +4,15 @@ On-demand CLI façade for Reyna family MCP services and LAN devices. The goal is
|
||||
|
||||
## Phase 2 scope
|
||||
|
||||
- **Devices parent command**: All local devices are now under `reyna-cli devices`.
|
||||
- **Devices parent command**: Non-Tactility local devices remain under `reyna-cli devices`; ESP32 boards use the dedicated `reyna-cli tactility` command group.
|
||||
- `reyna-cli devices screen ...` (ESP32 screen)
|
||||
- `reyna-cli devices laptop ...` (personal laptop MCP screen)
|
||||
- `reyna-cli devices computer ...` (this computer's desktop companion client)
|
||||
- `reyna-cli devices iphone ...` (iPhone app)
|
||||
- `reyna-cli devices arm ...` (Robot arm)
|
||||
- **Robot Arm**: Dynamic tool resolution for `state`, `home`, `wave`, `battery`.
|
||||
- **Tactility boards**: mDNS-first discovery with reserved-IP fallback, plus direct web API access for sysinfo, apps, app installation/run, and filesystem operations.
|
||||
- **Immich**: On-demand MCP bridge at `http://127.0.0.1:8626/mcp`.
|
||||
- **MongoDB**: On-demand MCP bridge at `http://127.0.0.1:8630/mcp`.
|
||||
- **Top-level aliases**: `reyna-cli screen`, `reyna-cli laptop`, `reyna-cli iphone`, `reyna-cli arm` still work for compatibility.
|
||||
- **Top-level aliases**: `reyna-cli screen`, `reyna-cli laptop`, and `reyna-cli iphone` still work for compatibility.
|
||||
|
||||
## Install / run
|
||||
|
||||
@@ -22,6 +21,23 @@ uv sync
|
||||
uv run reyna-cli doctor
|
||||
```
|
||||
|
||||
## Signed Python launcher (macOS permissions)
|
||||
|
||||
The manually built, signed **Reyna CLI.app** owns the macOS privacy identity. Use the `signed` wrapper to launch the current Python CLI through that app:
|
||||
|
||||
```bash
|
||||
# Optional: use a different active checkout or interpreter without rebuilding the app.
|
||||
cat > ~/.reyna-cli.env <<'EOF'
|
||||
REYNA_CLI_DIR=/Users/adolforeyna/Projects/platform/reyna-cli
|
||||
REYNA_CLI_PYTHON=/Users/adolforeyna/Projects/platform/reyna-cli/.venv/bin/python
|
||||
EOF
|
||||
|
||||
uv run reyna-cli signed doctor
|
||||
uv run reyna-cli signed local-services speech transcribe-file ./sample.wav --json
|
||||
```
|
||||
|
||||
The signed launcher reads `~/.reyna-cli.env`, then `~/.config/reyna-cli/env`; `REYNA_CLI_DIR` and `REYNA_CLI_PYTHON` environment variables override those values. Python-only edits take effect on the next `signed` run—**do not rebuild or re-sign the app for those edits**. Changes to Swift sources, `Info.plist`, or app signing require a new manual signed Xcode build before they can be used by `signed`.
|
||||
|
||||
## Device discovery
|
||||
|
||||
```bash
|
||||
@@ -29,6 +45,25 @@ uv run reyna-cli devices list --json
|
||||
uv run reyna-cli devices ping esp32_screen --json
|
||||
uv run reyna-cli devices tools esp32_screen --json
|
||||
uv run reyna-cli devices describe esp32_screen --for-hermes
|
||||
|
||||
# Tactility fleet discovery; offline boards remain visible in the JSON report
|
||||
uv run reyna-cli tactility discover --json
|
||||
uv run reyna-cli tactility sysinfo kidsos1 --json
|
||||
uv run reyna-cli tactility apps kidsos1 --json
|
||||
uv run reyna-cli tactility tools Grace --json
|
||||
uv run reyna-cli tactility describe Grace --json
|
||||
uv run reyna-cli tactility call Grace get_screenshot --args '{}' --json
|
||||
uv run reyna-cli tactility call Grace draw_color_bmp --args '{"bmp_base64":"...","x":0,"y":0}' --json
|
||||
uv run reyna-cli tactility call Grace play_tone --args '{"frequency":440,"duration_ms":500,"volume":40}' --json
|
||||
uv run reyna-cli tactility call Grace get_sensors --args '{}' --json
|
||||
uv run reyna-cli tactility install kidsos1 ./build/my.app --json
|
||||
uv run reyna-cli tactility run kidsos1 one.tactility.myapp --json
|
||||
uv run reyna-cli tactility report kidsos1 --json
|
||||
uv run reyna-cli tactility screen clear kidsos1 --width 320 --height 240 --json
|
||||
uv run reyna-cli tactility screen text kidsos1 "Grace\\nReady" --json
|
||||
uv run reyna-cli tactility screen text kidsos1 "Layer intentionally" --no-clear-first --json
|
||||
uv run reyna-cli tactility fs list kidsos1 --path /sdcard --json
|
||||
uv run reyna-cli tactility fs upload kidsos1 ./manifest.json /sdcard/manifest.json --json
|
||||
```
|
||||
|
||||
## Generic MCP calls
|
||||
@@ -37,15 +72,6 @@ uv run reyna-cli devices describe esp32_screen --for-hermes
|
||||
uv run reyna-cli devices call esp32_screen draw_text --args '{"text":"Hello","x":10,"y":20,"size":2}' --json
|
||||
```
|
||||
|
||||
## Robot Arm
|
||||
|
||||
```bash
|
||||
uv run reyna-cli devices arm state
|
||||
uv run reyna-cli devices arm home
|
||||
uv run reyna-cli devices arm wave
|
||||
uv run reyna-cli devices arm battery --json
|
||||
```
|
||||
|
||||
## ESP32 screen, laptop screen, this computer & iPhone
|
||||
|
||||
```bash
|
||||
|
||||
+133
-1
@@ -22,10 +22,11 @@ from reyna_cli.apple_llm_client import (
|
||||
from reyna_cli.deco_direct import DecoDirectClient
|
||||
from reyna_cli.desktop_service import SERVICE_NAME, service_action, service_status, unit_content, unit_path
|
||||
from reyna_cli.email_local import ThunderbirdEmailClient, email_tool_specs
|
||||
from reyna_cli.gitea_direct import GiteaClient, load_credentials
|
||||
from reyna_cli.immich import ImmichClient
|
||||
from reyna_cli.mcp import MCPClient
|
||||
from reyna_cli.mongo_direct import MongoDirectClient
|
||||
from reyna_cli.notes_direct import NotesAutomationError, create_note, list_notes, read_note
|
||||
from reyna_cli.notes_direct import NotesAutomationError, create_note, delete_note, list_notes, read_note, update_note
|
||||
from reyna_cli.privacy_client import default_socket_path as privacy_default_socket_path
|
||||
from reyna_cli.privacy_host import native_calendar_list, privacy_host_status_payload
|
||||
from reyna_cli.remarkable import LISTENER_LABEL, listen_forever, listener_service_action, listener_service_status, sync_once
|
||||
@@ -51,6 +52,7 @@ immich_app = typer.Typer(help="Immich direct REST API commands.")
|
||||
mongo_app = typer.Typer(help="MongoDB direct driver commands.")
|
||||
zoom_app = typer.Typer(help="Zoom direct REST API commands.")
|
||||
email_app = typer.Typer(help="Read-only local Thunderbird email commands.")
|
||||
gitea_app = typer.Typer(help="Direct Gitea API and credential diagnostics (never prints tokens).")
|
||||
deco_app = typer.Typer(help="TP-Link Deco direct router commands.")
|
||||
macmini_app = typer.Typer(help="Mac mini MCP tools (Calendar, Contacts, Reminders, Deco).")
|
||||
remarkable_app = typer.Typer(help="Paper Pro discovery, local cache, and macOS listener service.")
|
||||
@@ -79,6 +81,7 @@ app.add_typer(immich_app, name="immich")
|
||||
app.add_typer(mongo_app, name="mongo")
|
||||
app.add_typer(zoom_app, name="zoom")
|
||||
app.add_typer(email_app, name="email")
|
||||
app.add_typer(gitea_app, name="gitea")
|
||||
app.add_typer(deco_app, name="deco")
|
||||
macmini_app.add_typer(macmini_calendar_app, name="calendar")
|
||||
macmini_app.add_typer(macmini_contacts_app, name="contacts")
|
||||
@@ -142,6 +145,35 @@ def notes_create(
|
||||
fail(str(exc), json_output)
|
||||
|
||||
|
||||
@notes_app.command("edit")
|
||||
def notes_edit(
|
||||
note_id: str,
|
||||
title: Optional[str] = typer.Option(None, "--title"),
|
||||
body: Optional[str] = typer.Option(None, "--body"),
|
||||
json_output: bool = typer.Option(False, "--json"),
|
||||
):
|
||||
"""Replace the title and/or body of one Apple Note by ID."""
|
||||
try:
|
||||
emit({"ok": True, "source": "direct_apple_notes", "note": update_note(note_id, title=title, body=body)}, json_output)
|
||||
except (NotesAutomationError, ValueError) as exc:
|
||||
fail(str(exc), json_output)
|
||||
|
||||
|
||||
@notes_app.command("delete")
|
||||
def notes_delete(
|
||||
note_id: str,
|
||||
force: bool = typer.Option(False, "--force", help="Required: permanently delete the note."),
|
||||
json_output: bool = typer.Option(False, "--json"),
|
||||
):
|
||||
"""Permanently delete one Apple Note by ID; requires explicit --force."""
|
||||
if not force:
|
||||
fail("Deletion requires --force", json_output)
|
||||
try:
|
||||
emit({"ok": True, "source": "direct_apple_notes", "note": delete_note(note_id)}, json_output)
|
||||
except (NotesAutomationError, ValueError) as exc:
|
||||
fail(str(exc), json_output)
|
||||
|
||||
|
||||
def scrub_sensitive(value: Any) -> Any:
|
||||
if isinstance(value, dict):
|
||||
cleaned: Dict[str, Any] = {}
|
||||
@@ -170,6 +202,56 @@ def fail(message: str, json_output: bool = False, **extra: Any) -> None:
|
||||
raise typer.Exit(1)
|
||||
|
||||
|
||||
@gitea_app.command("credential")
|
||||
def gitea_credential(json_output: bool = typer.Option(False, "--json")):
|
||||
"""Report whether a Gitea credential is configured, without displaying it."""
|
||||
emit({"ok": True, **load_credentials().public_dict()}, json_output)
|
||||
|
||||
|
||||
@gitea_app.command("access")
|
||||
def gitea_access(json_output: bool = typer.Option(False, "--json")):
|
||||
"""Read the authenticated Gitea user endpoint."""
|
||||
try:
|
||||
emit({"ok": True, "user": GiteaClient().access()}, json_output)
|
||||
except Exception as exc:
|
||||
fail(str(exc), json_output)
|
||||
|
||||
|
||||
@gitea_app.command("repos")
|
||||
def gitea_repos(
|
||||
limit: int = typer.Option(50, min=1, max=100),
|
||||
json_output: bool = typer.Option(False, "--json"),
|
||||
):
|
||||
"""List repositories when the configured token has repository-list scope."""
|
||||
try:
|
||||
emit({"ok": True, "repos": GiteaClient().repos(limit)}, json_output)
|
||||
except Exception as exc:
|
||||
fail(str(exc), json_output)
|
||||
|
||||
|
||||
@gitea_app.command("repo")
|
||||
def gitea_repo(name: str, json_output: bool = typer.Option(False, "--json")):
|
||||
"""Read one repository and its permissions, e.g. adolforeyna/reyna-cli."""
|
||||
try:
|
||||
emit({"ok": True, "repo": GiteaClient().repo(name)}, json_output)
|
||||
except Exception as exc:
|
||||
fail(str(exc), json_output)
|
||||
|
||||
|
||||
def run_remindctl(args: List[str]) -> Any:
|
||||
"""Execute the supported Reminders CLI without invoking a shell."""
|
||||
result = subprocess.run(["remindctl", *args], capture_output=True, text=True, check=False)
|
||||
if result.returncode:
|
||||
raise RuntimeError(result.stderr.strip() or result.stdout.strip() or "remindctl command failed")
|
||||
output = result.stdout.strip()
|
||||
if not output:
|
||||
return {}
|
||||
try:
|
||||
return json.loads(output)
|
||||
except json.JSONDecodeError:
|
||||
return {"output": output}
|
||||
|
||||
|
||||
def get_required_device(name: str, json_output: bool = False) -> Device:
|
||||
device = get_device(name)
|
||||
if not device:
|
||||
@@ -2100,6 +2182,56 @@ def macmini_reminders_create(
|
||||
fail(str(exc), json_output)
|
||||
|
||||
|
||||
@macmini_reminders_app.command("edit")
|
||||
def macmini_reminders_edit(
|
||||
reminder_id: str,
|
||||
title: Optional[str] = typer.Option(None, "--title"),
|
||||
list_name: Optional[str] = typer.Option(None, "--list"),
|
||||
due: Optional[str] = typer.Option(None, "--due"),
|
||||
notes: Optional[str] = typer.Option(None, "--notes"),
|
||||
priority: Optional[str] = typer.Option(None, "--priority"),
|
||||
clear_due: bool = typer.Option(False, "--clear-due"),
|
||||
complete: bool = typer.Option(False, "--complete"),
|
||||
incomplete: bool = typer.Option(False, "--incomplete"),
|
||||
json_output: bool = typer.Option(False, "--json"),
|
||||
):
|
||||
"""Edit a reminder by its ID or ID prefix returned by ``reminders list``."""
|
||||
if complete and incomplete:
|
||||
fail("Use only one of --complete or --incomplete", json_output)
|
||||
if not any((title is not None, list_name is not None, due is not None, notes is not None, priority is not None, clear_due, complete, incomplete)):
|
||||
fail("Specify at least one field to edit", json_output)
|
||||
args = ["edit", reminder_id]
|
||||
for flag, value in (("--title", title), ("--list", list_name), ("--due", due), ("--notes", notes), ("--priority", priority)):
|
||||
if value is not None:
|
||||
args.extend([flag, value])
|
||||
if clear_due:
|
||||
args.append("--clear-due")
|
||||
if complete:
|
||||
args.append("--complete")
|
||||
if incomplete:
|
||||
args.append("--incomplete")
|
||||
args.extend(["--json", "--no-input"])
|
||||
try:
|
||||
emit({"ok": True, "source": "remindctl", "result": run_remindctl(args)}, json_output)
|
||||
except Exception as exc:
|
||||
fail(str(exc), json_output)
|
||||
|
||||
|
||||
@macmini_reminders_app.command("delete")
|
||||
def macmini_reminders_delete(
|
||||
reminder_ids: List[str] = typer.Argument(..., help="Reminder IDs or ID prefixes returned by reminders list."),
|
||||
force: bool = typer.Option(False, "--force", help="Required: permanently delete without an interactive prompt."),
|
||||
json_output: bool = typer.Option(False, "--json"),
|
||||
):
|
||||
"""Permanently delete one or more reminders; requires explicit --force."""
|
||||
if not force:
|
||||
fail("Deletion requires --force", json_output)
|
||||
try:
|
||||
emit({"ok": True, "source": "remindctl", "result": run_remindctl(["delete", *reminder_ids, "--force", "--json", "--no-input"])}, json_output)
|
||||
except Exception as exc:
|
||||
fail(str(exc), json_output)
|
||||
|
||||
|
||||
def emit_deco_result(tool: str, result: Dict[str, Any], json_output: bool = False) -> None:
|
||||
emit({"ok": True, "source": "direct", "tool": tool, "result": result}, json_output)
|
||||
|
||||
|
||||
+27
-3
@@ -1,6 +1,7 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import socket
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
@@ -36,6 +37,9 @@ class Device(BaseModel):
|
||||
urls.append(url)
|
||||
return urls
|
||||
|
||||
def resolved_host(self) -> str:
|
||||
return resolve_device_host(self)
|
||||
|
||||
def public_dict(self) -> Dict[str, Any]:
|
||||
return {
|
||||
"id": self.id,
|
||||
@@ -108,13 +112,12 @@ def load_registry(path: Optional[Path] = None) -> Registry:
|
||||
|
||||
def get_device(name: str, registry: Optional[Registry] = None) -> Optional[Device]:
|
||||
registry = registry or load_registry()
|
||||
normalized = name.strip()
|
||||
normalized = name.strip().casefold()
|
||||
shorthand = {
|
||||
"screen": "esp32_screen",
|
||||
"esp32": "esp32_screen",
|
||||
"iphone": "iphone_mcp",
|
||||
"phone": "iphone_mcp",
|
||||
"arm": "robot_arm",
|
||||
"laptop": "personal_laptop_screen",
|
||||
"laptop_screen": "personal_laptop_screen",
|
||||
"personal_laptop": "personal_laptop_screen",
|
||||
@@ -123,11 +126,32 @@ def get_device(name: str, registry: Optional[Registry] = None) -> Optional[Devic
|
||||
"local": "local_desktop_screen",
|
||||
}
|
||||
normalized = shorthand.get(normalized, normalized)
|
||||
aliases = {str(key).casefold(): value for key, value in registry.aliases.items()}
|
||||
normalized = aliases.get(normalized, normalized)
|
||||
for device in registry.devices:
|
||||
if normalized in {device.id, device.host, device.display_name}:
|
||||
candidates = {device.id.casefold(), device.host.casefold(), device.display_name.casefold()}
|
||||
if normalized in candidates:
|
||||
return device
|
||||
display = device.display_name.casefold()
|
||||
if display.startswith(normalized + " ") or display.startswith(normalized + "'s"):
|
||||
return device
|
||||
return None
|
||||
|
||||
|
||||
def resolve_device_host(device: Device) -> str:
|
||||
"""Resolve mDNS when available, retaining the registry IP for offline boards."""
|
||||
if device.host:
|
||||
try:
|
||||
return socket.gethostbyname(device.host)
|
||||
except OSError:
|
||||
pass
|
||||
return device.reserved_ip or device.host
|
||||
|
||||
|
||||
def is_tactility_device(device: Device) -> bool:
|
||||
value = f"{device.id} {device.type} {' '.join(device.capabilities)}".lower()
|
||||
return "tactility" in value or device.id.startswith("kidsos_") or device.id == "esp32_screen"
|
||||
|
||||
|
||||
def cache_path_for(device_id: str) -> Path:
|
||||
return CACHE_DIR / "devices" / f"{device_id}-tools.json"
|
||||
|
||||
@@ -0,0 +1,174 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
from urllib.parse import unquote, urlsplit
|
||||
|
||||
import httpx
|
||||
|
||||
from reyna_cli.env import load_hermes_env
|
||||
|
||||
DEFAULT_GITEA_URL = "https://git.reynafamily.com"
|
||||
SENSITIVE_KEYS = {"token", "access_token", "password", "authorization", "refresh_token"}
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class GiteaCredentials:
|
||||
username: str
|
||||
token: str
|
||||
source: str
|
||||
base_url: str
|
||||
|
||||
def public_dict(self) -> dict[str, Any]:
|
||||
parsed = urlsplit(self.base_url)
|
||||
return {
|
||||
"configured": bool(self.token) or self.source == "Pi 5 git credential helper",
|
||||
"host": parsed.netloc,
|
||||
"source": self.source,
|
||||
}
|
||||
|
||||
|
||||
def gitea_url() -> str:
|
||||
load_hermes_env()
|
||||
return os.getenv("REYNA_GITEA_URL") or os.getenv("GITEA_URL") or DEFAULT_GITEA_URL
|
||||
|
||||
|
||||
def _credential_from_file(path: Path, host: str) -> tuple[str, str] | None:
|
||||
try:
|
||||
lines = path.read_text(encoding="utf-8", errors="ignore").splitlines()
|
||||
except OSError:
|
||||
return None
|
||||
for line in lines:
|
||||
parsed = urlsplit(line.strip())
|
||||
if parsed.hostname == host and parsed.username and parsed.password:
|
||||
return unquote(parsed.username), unquote(parsed.password)
|
||||
return None
|
||||
|
||||
|
||||
def _credential_from_git_helper(host: str) -> tuple[str, str] | None:
|
||||
try:
|
||||
result = subprocess.run(
|
||||
["git", "credential", "fill"],
|
||||
input=f"protocol=https\nhost={host}\n\n",
|
||||
text=True,
|
||||
capture_output=True,
|
||||
timeout=10,
|
||||
check=False,
|
||||
)
|
||||
except (OSError, subprocess.TimeoutExpired):
|
||||
return None
|
||||
if result.returncode:
|
||||
return None
|
||||
values = dict(line.split("=", 1) for line in result.stdout.splitlines() if "=" in line)
|
||||
username, password = values.get("username", ""), values.get("password", "")
|
||||
return (username, password) if username and password else None
|
||||
|
||||
|
||||
def load_credentials() -> GiteaCredentials:
|
||||
load_hermes_env()
|
||||
base_url = gitea_url().rstrip("/")
|
||||
host = urlsplit(base_url).hostname or ""
|
||||
token = os.getenv("REYNA_GITEA_TOKEN") or os.getenv("GITEA_TOKEN")
|
||||
if token:
|
||||
return GiteaCredentials(os.getenv("REYNA_GITEA_USERNAME", "adolforeyna"), token, "REYNA_GITEA_TOKEN", base_url)
|
||||
|
||||
credentials_path = Path(os.getenv("GIT_CREDENTIALS_FILE", Path.home() / ".git-credentials"))
|
||||
found = _credential_from_file(credentials_path, host)
|
||||
if found:
|
||||
return GiteaCredentials(*found, "GIT_CREDENTIALS_FILE", base_url)
|
||||
|
||||
found = _credential_from_git_helper(host)
|
||||
if found:
|
||||
return GiteaCredentials(*found, "git credential helper", base_url)
|
||||
return GiteaCredentials("", "", "Pi 5 git credential helper", base_url)
|
||||
|
||||
|
||||
def redact_sensitive(value: Any) -> Any:
|
||||
if isinstance(value, dict):
|
||||
return {key: "[REDACTED]" if key.casefold() in SENSITIVE_KEYS else redact_sensitive(item) for key, item in value.items()}
|
||||
if isinstance(value, list):
|
||||
return [redact_sensitive(item) for item in value]
|
||||
return value
|
||||
|
||||
|
||||
class GiteaClient:
|
||||
def __init__(self, credentials: GiteaCredentials | None = None, timeout: float = 20.0, transport: httpx.BaseTransport | None = None):
|
||||
self.credentials = credentials or load_credentials()
|
||||
self.remote_credential_helper = self.credentials.source == "Pi 5 git credential helper"
|
||||
if not self.credentials.token and not self.remote_credential_helper:
|
||||
raise RuntimeError("No Gitea credential is available. Set REYNA_GITEA_TOKEN or configure a supported Git credential source.")
|
||||
self.client = httpx.Client(
|
||||
base_url=self.credentials.base_url,
|
||||
timeout=timeout,
|
||||
transport=transport,
|
||||
headers={"Authorization": f"token {self.credentials.token}", "Accept": "application/json"},
|
||||
)
|
||||
|
||||
def _remote_get(self, path: str) -> Any:
|
||||
"""Use Pi 5's existing Git credential helper without returning its token."""
|
||||
script = r'''import json, subprocess, sys, urllib.request
|
||||
from urllib.parse import urlsplit
|
||||
base_url, path = sys.argv[1:]
|
||||
host = urlsplit(base_url).hostname
|
||||
result = subprocess.run(["git", "credential", "fill"], input=f"protocol=https\nhost={host}\n\n", text=True, capture_output=True, check=False)
|
||||
values = dict(line.split("=", 1) for line in result.stdout.splitlines() if "=" in line)
|
||||
token = values.get("password", "")
|
||||
if result.returncode or not token:
|
||||
raise SystemExit("Pi 5 Git credential helper did not provide a Gitea credential")
|
||||
request = urllib.request.Request(base_url.rstrip("/") + path, headers={"Authorization": "token " + token, "Accept": "application/json"})
|
||||
with urllib.request.urlopen(request, timeout=20) as response:
|
||||
payload = json.load(response)
|
||||
def redact(value):
|
||||
if isinstance(value, dict):
|
||||
return {key: "[REDACTED]" if key.casefold() in {"token", "access_token", "password", "authorization", "refresh_token"} else redact(item) for key, item in value.items()}
|
||||
if isinstance(value, list):
|
||||
return [redact(item) for item in value]
|
||||
return value
|
||||
print(json.dumps(redact(payload)))'''
|
||||
helper_host = os.getenv("REYNA_GITEA_CREDENTIAL_HOST", "pi5")
|
||||
encoded_script = base64.b64encode(script.encode("utf-8")).decode("ascii")
|
||||
remote_command = (
|
||||
"python3 -c \"import base64;exec(base64.b64decode('"
|
||||
+ encoded_script
|
||||
+ "'))\" "
|
||||
+ self.credentials.base_url
|
||||
+ " "
|
||||
+ path
|
||||
)
|
||||
result = subprocess.run(
|
||||
["ssh", "-o", "BatchMode=yes", helper_host, remote_command],
|
||||
text=True,
|
||||
capture_output=True,
|
||||
timeout=30,
|
||||
check=False,
|
||||
)
|
||||
if result.returncode:
|
||||
raise RuntimeError("Pi 5 Gitea credential helper request failed.")
|
||||
try:
|
||||
return json.loads(result.stdout)
|
||||
except json.JSONDecodeError as exc:
|
||||
raise RuntimeError("Pi 5 Gitea credential helper returned invalid JSON.") from exc
|
||||
|
||||
def _get(self, path: str) -> Any:
|
||||
if self.remote_credential_helper:
|
||||
return self._remote_get(path)
|
||||
response = self.client.get(path)
|
||||
response.raise_for_status()
|
||||
return redact_sensitive(response.json())
|
||||
|
||||
def access(self) -> Any:
|
||||
return self._get("/api/v1/user")
|
||||
|
||||
def repos(self, limit: int = 50) -> Any:
|
||||
return self._get("/api/v1/user/repos?limit=" + str(limit))
|
||||
|
||||
def repo(self, name: str) -> Any:
|
||||
owner, separator, repo = name.partition("/")
|
||||
if not separator or not owner or not repo:
|
||||
raise ValueError("Repository must be in owner/name form.")
|
||||
return self._get(f"/api/v1/repos/{owner}/{repo}")
|
||||
@@ -105,6 +105,53 @@ _NOTES_CREATE_SCRIPT = r'''function run(argv) {
|
||||
}'''
|
||||
|
||||
|
||||
_NOTES_UPDATE_SCRIPT = r'''function run(argv) {
|
||||
const input = JSON.parse(argv[0]);
|
||||
const app = Application("/System/Applications/Notes.app");
|
||||
const escapeHtml = value => String(value).replace(/&/g, "&").replace(/</g, "<").replace(/>/g, ">").replace(/"/g, """);
|
||||
const accounts = app.accounts();
|
||||
for (let a = 0; a < accounts.length; a++) {
|
||||
const folders = accounts[a].folders();
|
||||
for (let f = 0; f < folders.length; f++) {
|
||||
const notes = folders[f].notes();
|
||||
for (let n = 0; n < notes.length; n++) {
|
||||
const note = notes[n];
|
||||
if (String(note.id()) !== input.id) continue;
|
||||
const oldTitle = String(note.name());
|
||||
const oldText = String(note.plaintext());
|
||||
const fallbackBody = oldText.startsWith(oldTitle) ? oldText.slice(oldTitle.length).replace(/^\\n+/, "") : oldText;
|
||||
const title = input.title === null ? oldTitle : input.title;
|
||||
const body = input.body === null ? fallbackBody : input.body;
|
||||
note.body = "<h1>" + escapeHtml(title) + "</h1><div>" + escapeHtml(body).replace(/\\n/g, "<br>") + "</div>";
|
||||
return JSON.stringify({id: String(note.id()), title: String(note.name()), folder: String(folders[f].name())});
|
||||
}
|
||||
}
|
||||
}
|
||||
throw new Error("Note not found");
|
||||
}'''
|
||||
|
||||
|
||||
_NOTES_DELETE_SCRIPT = r'''function run(argv) {
|
||||
const input = JSON.parse(argv[0]);
|
||||
const app = Application("/System/Applications/Notes.app");
|
||||
const accounts = app.accounts();
|
||||
for (let a = 0; a < accounts.length; a++) {
|
||||
const folders = accounts[a].folders();
|
||||
for (let f = 0; f < folders.length; f++) {
|
||||
const notes = folders[f].notes();
|
||||
for (let n = 0; n < notes.length; n++) {
|
||||
const note = notes[n];
|
||||
if (String(note.id()) !== input.id) continue;
|
||||
const id = String(note.id());
|
||||
note.delete();
|
||||
return JSON.stringify({id: id, deleted: true});
|
||||
}
|
||||
}
|
||||
}
|
||||
throw new Error("Note not found");
|
||||
}'''
|
||||
|
||||
|
||||
def _run_jxa(script: str, payload: dict[str, Any], *, runner: Runner = subprocess.run) -> Any:
|
||||
arguments: Sequence[str] = [
|
||||
"/usr/bin/osascript",
|
||||
@@ -178,3 +225,30 @@ def create_note(
|
||||
if not isinstance(result, dict):
|
||||
raise NotesAutomationError("Apple Notes create response was not an object")
|
||||
return result
|
||||
|
||||
|
||||
def update_note(
|
||||
note_id: str,
|
||||
*,
|
||||
title: Optional[str] = None,
|
||||
body: Optional[str] = None,
|
||||
runner: Runner = subprocess.run,
|
||||
) -> dict[str, Any]:
|
||||
note_id = _bounded_text(note_id, "note id", 1024)
|
||||
if title is None and body is None:
|
||||
raise ValueError("Specify title or body to update")
|
||||
if title is not None:
|
||||
title = _bounded_text(title, "title", 500)
|
||||
if body is not None and len(body) > 100_000:
|
||||
raise ValueError("body exceeds maximum length 100000")
|
||||
result = _run_jxa(_NOTES_UPDATE_SCRIPT, {"id": note_id, "title": title, "body": body}, runner=runner)
|
||||
if not isinstance(result, dict):
|
||||
raise NotesAutomationError("Apple Notes update response was not an object")
|
||||
return result
|
||||
|
||||
|
||||
def delete_note(note_id: str, *, runner: Runner = subprocess.run) -> dict[str, Any]:
|
||||
result = _run_jxa(_NOTES_DELETE_SCRIPT, {"id": _bounded_text(note_id, "note id", 1024)}, runner=runner)
|
||||
if not isinstance(result, dict) or result.get("deleted") is not True:
|
||||
raise NotesAutomationError("Apple Notes delete response was not confirmed")
|
||||
return result
|
||||
|
||||
@@ -0,0 +1,123 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, Optional
|
||||
from urllib.parse import quote
|
||||
|
||||
import httpx
|
||||
from PIL import Image, ImageDraw, ImageFont
|
||||
|
||||
|
||||
class TactilityClient:
|
||||
"""Small direct client for the Tactility firmware HTTP API."""
|
||||
|
||||
def __init__(self, base_url: str, timeout: float = 30.0, transport: Optional[httpx.BaseTransport] = None):
|
||||
self.base_url = base_url.rstrip("/")
|
||||
self.client = httpx.Client(timeout=timeout, transport=transport, follow_redirects=True)
|
||||
|
||||
def _response(self, response: httpx.Response) -> Any:
|
||||
response.raise_for_status()
|
||||
if not response.content:
|
||||
return {"ok": True, "status_code": response.status_code}
|
||||
content_type = response.headers.get("content-type", "")
|
||||
if "json" in content_type:
|
||||
return response.json()
|
||||
try:
|
||||
return response.json()
|
||||
except ValueError:
|
||||
return {"ok": True, "status_code": response.status_code, "response": response.text}
|
||||
|
||||
def get(self, path: str, **kwargs: Any) -> Any:
|
||||
return self._response(self.client.get(f"{self.base_url}{path}", **kwargs))
|
||||
|
||||
def post(self, path: str, **kwargs: Any) -> Any:
|
||||
return self._response(self.client.post(f"{self.base_url}{path}", **kwargs))
|
||||
|
||||
def sysinfo(self) -> Any:
|
||||
return self.get("/api/sysinfo")
|
||||
|
||||
def apps(self) -> Any:
|
||||
return self.get("/api/apps")
|
||||
|
||||
def install_app(self, app_path: Path) -> Any:
|
||||
with app_path.open("rb") as handle:
|
||||
response = self.client.put(
|
||||
f"{self.base_url}/api/apps/install",
|
||||
files={"file": (app_path.name, handle, "application/octet-stream")},
|
||||
)
|
||||
return self._response(response)
|
||||
|
||||
def run_app(self, app_id: str) -> Any:
|
||||
return self.post("/api/apps/run", params={"id": app_id})
|
||||
|
||||
def fs_list(self, path: str = "/") -> Any:
|
||||
return self.get("/fs/list", params={"path": path})
|
||||
|
||||
def fs_mkdir(self, path: str) -> Any:
|
||||
return self.post("/fs/mkdir", params={"path": path})
|
||||
|
||||
def fs_upload(self, local_path: Path, remote_path: str) -> Any:
|
||||
data = local_path.read_bytes()
|
||||
return self.post(
|
||||
"/fs/upload",
|
||||
params={"path": remote_path},
|
||||
content=data,
|
||||
headers={"Content-Type": "application/octet-stream", "Content-Length": str(len(data))},
|
||||
)
|
||||
|
||||
def fs_download(self, remote_path: str, local_path: Path) -> Dict[str, Any]:
|
||||
response = self.client.get(f"{self.base_url}/fs/download", params={"path": remote_path})
|
||||
response.raise_for_status()
|
||||
local_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
local_path.write_bytes(response.content)
|
||||
return {"ok": True, "path": str(local_path), "bytes": len(response.content)}
|
||||
|
||||
def fs_read(self, remote_path: str) -> str:
|
||||
response = self.client.get(f"{self.base_url}/fs/download", params={"path": remote_path})
|
||||
response.raise_for_status()
|
||||
return response.content.decode("utf-8", errors="replace")
|
||||
|
||||
def fs_delete(self, path: str) -> Any:
|
||||
return self.post("/fs/delete", params={"path": path})
|
||||
|
||||
def fs_rename(self, path: str, new_name: str) -> Any:
|
||||
return self.post("/fs/rename", params={"path": path, "newName": new_name})
|
||||
|
||||
def screen_raw(self, pixels: bytes, width: int, height: int) -> Any:
|
||||
expected = width * height * 2
|
||||
if len(pixels) != expected:
|
||||
raise ValueError(f"RGB565 frame is {len(pixels)} bytes; expected {expected} for {width}x{height}")
|
||||
response = self.client.post(
|
||||
f"{self.base_url}/api/screen/raw",
|
||||
params={"w": width, "h": height},
|
||||
content=pixels,
|
||||
headers={"Content-Type": "application/octet-stream", "Content-Length": str(len(pixels))},
|
||||
)
|
||||
return self._response(response)
|
||||
|
||||
def screen_clear_raw(self, width: int, height: int) -> Any:
|
||||
return self.screen_raw(bytes(width * height * 2), width, height)
|
||||
|
||||
@staticmethod
|
||||
def _rgb565(image: Image.Image) -> bytes:
|
||||
pixels = bytearray()
|
||||
for red, green, blue in image.convert("RGB").get_flattened_data():
|
||||
pixels.extend((((red & 0xF8) << 8) | ((green & 0xFC) << 3) | (blue >> 3)).to_bytes(2, "big"))
|
||||
return bytes(pixels)
|
||||
|
||||
def screen_text(self, message: str, width: int, height: int, clear_first: bool = True) -> Dict[str, Any]:
|
||||
clear_result = self.screen_clear_raw(width, height) if clear_first else None
|
||||
image = Image.new("RGB", (width, height), (14, 18, 38))
|
||||
draw = ImageDraw.Draw(image)
|
||||
font_path = next((path for path in (
|
||||
"/System/Library/Fonts/SFNS.ttf",
|
||||
"/Library/Fonts/Arial.ttf",
|
||||
"/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf",
|
||||
) if Path(path).exists()), None)
|
||||
font_size = max(16, round(width / 16))
|
||||
font = ImageFont.truetype(font_path, font_size) if font_path else ImageFont.load_default()
|
||||
margin = max(1, min(8, min(width, height) // 10))
|
||||
draw.rectangle((margin, margin, width - margin - 1, height - margin - 1), outline=(90, 110, 180), width=2)
|
||||
draw.multiline_text((width // 2, height // 2), message, font=font, fill=(245, 245, 232), anchor="mm", align="center", spacing=5)
|
||||
write_result = self.screen_raw(self._rgb565(image), width, height)
|
||||
return {"clear_first": clear_first, "clear": clear_result, "write": write_result, "width": width, "height": height}
|
||||
@@ -0,0 +1,84 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
import httpx
|
||||
|
||||
from reyna_cli.gitea_direct import GiteaClient, GiteaCredentials, load_credentials, redact_sensitive
|
||||
|
||||
|
||||
def test_load_credentials_prefers_environment_without_exposing_token(monkeypatch):
|
||||
monkeypatch.setenv("REYNA_GITEA_TOKEN", "test-token-value")
|
||||
monkeypatch.setenv("REYNA_GITEA_URL", "https://git.example.test/")
|
||||
|
||||
credentials = load_credentials()
|
||||
|
||||
assert credentials.token == "test-token-value"
|
||||
assert credentials.source == "REYNA_GITEA_TOKEN"
|
||||
assert credentials.public_dict() == {
|
||||
"configured": True,
|
||||
"host": "git.example.test",
|
||||
"source": "REYNA_GITEA_TOKEN",
|
||||
}
|
||||
assert "token" not in credentials.public_dict()
|
||||
|
||||
|
||||
def test_load_credentials_reads_git_credentials_file(monkeypatch, tmp_path: Path):
|
||||
credentials_file = tmp_path / "credentials"
|
||||
credentials_file.write_text(
|
||||
"https://adolforeyna:stored-secret@git.reynafamily.com\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
monkeypatch.delenv("REYNA_GITEA_TOKEN", raising=False)
|
||||
monkeypatch.delenv("GITEA_TOKEN", raising=False)
|
||||
monkeypatch.setenv("GIT_CREDENTIALS_FILE", str(credentials_file))
|
||||
|
||||
credentials = load_credentials()
|
||||
|
||||
assert credentials.username == "adolforeyna"
|
||||
assert credentials.token == "stored-secret"
|
||||
assert credentials.source == "GIT_CREDENTIALS_FILE"
|
||||
assert "stored-secret" not in str(credentials.public_dict())
|
||||
|
||||
|
||||
def test_repo_response_is_redacted_and_requests_authenticated_endpoint():
|
||||
requests: list[httpx.Request] = []
|
||||
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
requests.append(request)
|
||||
return httpx.Response(
|
||||
200,
|
||||
json={
|
||||
"full_name": "adolforeyna/reyna-cli",
|
||||
"permissions": {"push": True},
|
||||
"token": "response-secret",
|
||||
},
|
||||
)
|
||||
|
||||
client = GiteaClient(
|
||||
GiteaCredentials("adolforeyna", "request-secret", "test", "https://git.example.test"),
|
||||
transport=httpx.MockTransport(handler),
|
||||
)
|
||||
|
||||
result = client.repo("adolforeyna/reyna-cli")
|
||||
|
||||
assert requests[0].url.path == "/api/v1/repos/adolforeyna/reyna-cli"
|
||||
assert requests[0].headers["authorization"] == "token request-secret"
|
||||
assert result["permissions"]["push"] is True
|
||||
assert result["token"] == "[REDACTED]"
|
||||
assert "response-secret" not in str(result)
|
||||
|
||||
|
||||
def test_remote_credential_helper_can_make_a_safe_api_request(monkeypatch):
|
||||
credentials = GiteaCredentials("", "", "Pi 5 git credential helper", "https://git.example.test")
|
||||
assert credentials.public_dict()["configured"] is True
|
||||
client = GiteaClient(credentials)
|
||||
monkeypatch.setattr(client, "_remote_get", lambda path: {"full_name": "adolforeyna/reyna-cli"})
|
||||
|
||||
assert client.repo("adolforeyna/reyna-cli") == {"full_name": "adolforeyna/reyna-cli"}
|
||||
|
||||
|
||||
def test_redact_sensitive_handles_nested_api_values():
|
||||
assert redact_sensitive(
|
||||
{"password": "one", "nested": [{"access_token": "two"}], "name": "safe"}
|
||||
) == {"password": "[REDACTED]", "nested": [{"access_token": "[REDACTED]"}], "name": "safe"}
|
||||
@@ -12,7 +12,7 @@ from types import SimpleNamespace
|
||||
from typer.testing import CliRunner
|
||||
|
||||
from reyna_cli.cli import app
|
||||
from reyna_cli.notes_direct import create_note, list_notes, read_note
|
||||
from reyna_cli.notes_direct import create_note, delete_note, list_notes, read_note, update_note
|
||||
|
||||
runner = CliRunner()
|
||||
|
||||
@@ -63,6 +63,25 @@ def test_read_and_create_notes_use_json_arguments_without_script_interpolation()
|
||||
assert created["id"] == "new-1"
|
||||
|
||||
|
||||
def test_update_and_delete_notes_use_fixed_jxa_and_json_arguments():
|
||||
updated = update_note(
|
||||
"x-coredata://note-1",
|
||||
title="Updated",
|
||||
body="New body",
|
||||
runner=_success_runner(
|
||||
{"id": "x-coredata://note-1", "title": "Updated", "body": "New body"},
|
||||
{"id": "x-coredata://note-1", "title": "Updated"},
|
||||
),
|
||||
)
|
||||
assert updated["title"] == "Updated"
|
||||
|
||||
deleted = delete_note(
|
||||
"x-coredata://note-1",
|
||||
runner=_success_runner({"id": "x-coredata://note-1"}, {"id": "x-coredata://note-1", "deleted": True}),
|
||||
)
|
||||
assert deleted["deleted"] is True
|
||||
|
||||
|
||||
def test_notes_cli_is_available_at_top_level_and_macmini_alias(monkeypatch):
|
||||
monkeypatch.setattr("reyna_cli.cli.list_notes", lambda **_: [{"id": "n1", "title": "Test"}])
|
||||
|
||||
|
||||
@@ -248,6 +248,42 @@ def test_cli_reminders_no_mcp_tool_call_remaining():
|
||||
assert 'call_macmini_tool("reminders_create"' not in src
|
||||
|
||||
|
||||
def test_cli_reminders_edit_uses_remindctl(monkeypatch):
|
||||
captured = {}
|
||||
|
||||
def fake_run(args):
|
||||
captured["args"] = args
|
||||
return {"updated": {"id": "r1", "title": "New title"}}
|
||||
|
||||
monkeypatch.setattr("reyna_cli.cli.run_remindctl", fake_run)
|
||||
res = runner.invoke(app, ["macmini", "reminders", "edit", "r1", "--title", "New title", "--due", "tomorrow", "--json"])
|
||||
|
||||
assert res.exit_code == 0
|
||||
assert captured["args"] == ["edit", "r1", "--title", "New title", "--due", "tomorrow", "--json", "--no-input"]
|
||||
assert json.loads(res.stdout)["source"] == "remindctl"
|
||||
|
||||
|
||||
def test_cli_reminders_delete_requires_force(monkeypatch):
|
||||
res = runner.invoke(app, ["macmini", "reminders", "delete", "r1", "--json"])
|
||||
|
||||
assert res.exit_code == 1
|
||||
assert "--force" in res.stdout
|
||||
|
||||
|
||||
def test_cli_reminders_delete_uses_remindctl_with_force(monkeypatch):
|
||||
captured = {}
|
||||
|
||||
def fake_run(args):
|
||||
captured["args"] = args
|
||||
return {"deleted": ["r1"]}
|
||||
|
||||
monkeypatch.setattr("reyna_cli.cli.run_remindctl", fake_run)
|
||||
res = runner.invoke(app, ["macmini", "reminders", "delete", "r1", "--force", "--json"])
|
||||
|
||||
assert res.exit_code == 0
|
||||
assert captured["args"] == ["delete", "r1", "--force", "--json", "--no-input"]
|
||||
|
||||
|
||||
def test_privacy_host_cli_has_reminders_authorize():
|
||||
res = runner.invoke(app, ["privacy-host", "--help"])
|
||||
assert res.exit_code == 0
|
||||
|
||||
@@ -0,0 +1,113 @@
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
import httpx
|
||||
from typer.testing import CliRunner
|
||||
|
||||
from reyna_cli.cli import app
|
||||
from reyna_cli.config import Device, Registry, get_device, resolve_device_host
|
||||
from reyna_cli.tactility import TactilityClient
|
||||
|
||||
runner = CliRunner()
|
||||
|
||||
|
||||
def test_friendly_name_lookup_is_case_insensitive():
|
||||
registry = Registry(devices=[Device(id="kidsos_5c5c", display_name="Grace's 2.8-inch Tactility Board")])
|
||||
assert get_device("Grace", registry).id == "kidsos_5c5c"
|
||||
assert get_device("grace", registry).id == "kidsos_5c5c"
|
||||
|
||||
|
||||
def test_resolve_device_host_prefers_live_mdns(monkeypatch):
|
||||
monkeypatch.setattr("reyna_cli.config.socket.gethostbyname", lambda host: "192.168.68.141")
|
||||
device = Device(id="kidsos_1234", host="kidsos-1234.local", reserved_ip="192.168.68.99")
|
||||
assert resolve_device_host(device) == "192.168.68.141"
|
||||
|
||||
|
||||
def test_resolve_device_host_falls_back_when_mdns_is_offline(monkeypatch):
|
||||
def fail(_host):
|
||||
raise OSError("offline")
|
||||
|
||||
monkeypatch.setattr("reyna_cli.config.socket.gethostbyname", fail)
|
||||
device = Device(id="kidsos_1234", host="kidsos-1234.local", reserved_ip="192.168.68.99")
|
||||
assert resolve_device_host(device) == "192.168.68.99"
|
||||
|
||||
|
||||
def test_tactility_client_uses_web_api(monkeypatch, tmp_path):
|
||||
requests = []
|
||||
|
||||
def handler(request: httpx.Request):
|
||||
requests.append(request)
|
||||
if request.url.path == "/api/sysinfo":
|
||||
return httpx.Response(200, json={"version": "0.8.0-dev"})
|
||||
if request.url.path == "/api/apps":
|
||||
return httpx.Response(200, json={"apps": [{"id": "one.tactility.demo"}]})
|
||||
if request.url.path == "/fs/list":
|
||||
return httpx.Response(200, json={"path": "/sdcard", "entries": []})
|
||||
return httpx.Response(200, json={"ok": True})
|
||||
|
||||
client = TactilityClient("http://kidsos-1234.local", transport=httpx.MockTransport(handler))
|
||||
assert client.sysinfo()["version"] == "0.8.0-dev"
|
||||
assert client.apps()["apps"][0]["id"] == "one.tactility.demo"
|
||||
assert client.fs_list("/sdcard")["path"] == "/sdcard"
|
||||
assert requests[0].url.host == "kidsos-1234.local"
|
||||
|
||||
|
||||
def test_tactility_install_and_run_requests(tmp_path):
|
||||
requests = []
|
||||
|
||||
def handler(request: httpx.Request):
|
||||
requests.append(request)
|
||||
return httpx.Response(200, json={"ok": True})
|
||||
|
||||
app_file = tmp_path / "demo.app"
|
||||
app_file.write_bytes(b"APP")
|
||||
client = TactilityClient("http://192.168.68.99", transport=httpx.MockTransport(handler))
|
||||
client.install_app(app_file)
|
||||
client.run_app("one.tactility.demo")
|
||||
|
||||
assert requests[0].method == "PUT"
|
||||
assert requests[0].url.path == "/api/apps/install"
|
||||
assert requests[1].url.path == "/api/apps/run"
|
||||
assert requests[1].url.params["id"] == "one.tactility.demo"
|
||||
|
||||
|
||||
def test_tactility_screen_text_clears_before_write():
|
||||
requests = []
|
||||
|
||||
def handler(request: httpx.Request):
|
||||
requests.append(request)
|
||||
return httpx.Response(200, json={"ok": True})
|
||||
|
||||
client = TactilityClient("http://192.168.68.99", transport=httpx.MockTransport(handler))
|
||||
result = client.screen_text("Grace", 20, 10, clear_first=True)
|
||||
|
||||
assert result["clear_first"] is True
|
||||
assert [request.url.path for request in requests] == ["/api/screen/raw", "/api/screen/raw"]
|
||||
assert requests[0].content == bytes(20 * 10 * 2)
|
||||
assert len(requests[1].content) == 20 * 10 * 2
|
||||
|
||||
|
||||
def test_tactility_cli_has_commands(monkeypatch, tmp_path):
|
||||
registry_path = tmp_path / "devices.yaml"
|
||||
registry_path.write_text(
|
||||
"""
|
||||
devices:
|
||||
kidsos_1234:
|
||||
type: esp32-s3-tactility
|
||||
host: kidsos-1234.local
|
||||
reserved_ip: 192.168.68.99
|
||||
""",
|
||||
encoding="utf-8",
|
||||
)
|
||||
monkeypatch.setenv("REYNA_DEVICES_REGISTRY", str(registry_path))
|
||||
|
||||
result = runner.invoke(app, ["tactility", "--help"])
|
||||
assert result.exit_code == 0
|
||||
for command in ("discover", "sysinfo", "apps", "install", "run", "report", "fs", "screen"):
|
||||
assert command in result.stdout
|
||||
|
||||
|
||||
def test_robot_arm_is_not_a_cli_surface():
|
||||
result = runner.invoke(app, ["devices", "--help"])
|
||||
assert result.exit_code == 0
|
||||
assert "arm" not in result.stdout.lower()
|
||||
Reference in New Issue
Block a user