Adolfo Reyna
9fd04b0ce4
Add the owner-only AF_UNIX Reyna CLI privacy host, strict signed-app installation, and typed native routing for Calendar, Contacts, and Reminders.\n\nAdd bounded system-status paths and config-only direct local-service wrappers. Preserve MacMiniMCP pending explicit cutover approval.\n\nApple Notes is intentionally deferred: no native Notes operations, Apple Events declaration, or Automation helper are included; legacy Notes handling remains untouched.
245 lines
8.3 KiB
Python
245 lines
8.3 KiB
Python
"""Tests for PrivacyClient — TDD with real Unix socket servers, no socket mocks."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import socket
|
|
import threading
|
|
import time
|
|
import uuid
|
|
from pathlib import Path
|
|
import tempfile
|
|
|
|
import pytest
|
|
|
|
from reyna_cli.privacy_client import (
|
|
PrivacyClient,
|
|
PrivacyClientError,
|
|
default_socket_path,
|
|
)
|
|
|
|
|
|
def test_default_socket_path():
|
|
p = default_socket_path()
|
|
# must be Path and match ~/Library/Application Support/reyna-cli/privacy/reyna-cli.sock
|
|
expected = Path.home() / "Library/Application Support/reyna-cli/privacy/reyna-cli.sock"
|
|
# Also accept expanded: Path.home() / "Library/Application Support/reyna-cli/privacy/reyna-cli.sock"
|
|
# Compare string representation or Path equality
|
|
assert isinstance(p, Path)
|
|
assert p == expected
|
|
assert str(p).endswith("Library/Application Support/reyna-cli/privacy/reyna-cli.sock")
|
|
|
|
|
|
# --- helpers for real socket server ---
|
|
|
|
class OneShotServer:
|
|
"""Simple real Unix socket server that handles one connection with a custom handler."""
|
|
|
|
def __init__(self, handler):
|
|
self.handler = handler
|
|
self.tmpdir = tempfile.TemporaryDirectory()
|
|
self.sock_path = Path(self.tmpdir.name) / "test.sock"
|
|
self._thread = None
|
|
self._ready = threading.Event()
|
|
self._done = threading.Event()
|
|
self.exception = None
|
|
|
|
def start(self):
|
|
def run():
|
|
srv = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
|
|
try:
|
|
srv.bind(str(self.sock_path))
|
|
srv.listen(1)
|
|
self._ready.set()
|
|
srv.settimeout(5)
|
|
try:
|
|
conn, _ = srv.accept()
|
|
except socket.timeout:
|
|
return
|
|
try:
|
|
self.handler(conn)
|
|
except Exception as e:
|
|
self.exception = e
|
|
finally:
|
|
try:
|
|
conn.close()
|
|
except Exception:
|
|
pass
|
|
finally:
|
|
srv.close()
|
|
self._done.set()
|
|
|
|
self._thread = threading.Thread(target=run, daemon=True)
|
|
self._thread.start()
|
|
assert self._ready.wait(timeout=3), "server failed to start"
|
|
return self
|
|
|
|
def stop(self):
|
|
self._done.wait(timeout=3)
|
|
if self._thread:
|
|
self._thread.join(timeout=1)
|
|
self.tmpdir.cleanup()
|
|
if self.exception:
|
|
raise self.exception
|
|
|
|
def __enter__(self):
|
|
return self.start()
|
|
|
|
def __exit__(self, *args):
|
|
self.stop()
|
|
|
|
|
|
def read_one_line(conn: socket.socket, timeout=2) -> dict:
|
|
conn.settimeout(timeout)
|
|
buf = b""
|
|
while b"\n" not in buf:
|
|
chunk = conn.recv(4096)
|
|
if not chunk:
|
|
break
|
|
buf += chunk
|
|
line = buf.split(b"\n")[0]
|
|
return json.loads(line.decode("utf-8"))
|
|
|
|
|
|
def test_call_happy_path_sends_one_json_line_and_validates():
|
|
received = {}
|
|
|
|
def handler(conn):
|
|
# read exactly one json line
|
|
conn.settimeout(2)
|
|
data = b""
|
|
while not data.endswith(b"\n"):
|
|
chunk = conn.recv(4096)
|
|
if not chunk:
|
|
break
|
|
data += chunk
|
|
# ensure only one line sent (count newline)
|
|
if data.count(b"\n") > 1:
|
|
raise AssertionError("client sent more than one line")
|
|
assert data.endswith(b"\n")
|
|
obj = json.loads(data.decode())
|
|
received.update(obj)
|
|
assert "id" in obj and isinstance(obj["id"], str) and obj["id"]
|
|
assert obj["operation"] == "service.health"
|
|
assert obj["arguments"] == {"x": 1}
|
|
# echo back with same id
|
|
resp = {"id": obj["id"], "ok": True, "result": {"status": "ok"}}
|
|
conn.sendall((json.dumps(resp) + "\n").encode())
|
|
|
|
with OneShotServer(handler) as srv:
|
|
client = PrivacyClient(socket_path=srv.sock_path, timeout=2)
|
|
resp = client.call("service.health", {"x": 1})
|
|
|
|
assert resp["ok"] is True
|
|
assert resp["result"]["status"] == "ok"
|
|
assert received["id"]
|
|
# id uniqueness check - call again should be different
|
|
second_id = {}
|
|
|
|
def handler2(conn):
|
|
obj = read_one_line(conn)
|
|
second_id["id"] = obj["id"]
|
|
resp = {"id": obj["id"], "ok": True, "result": {}}
|
|
conn.sendall((json.dumps(resp) + "\n").encode())
|
|
|
|
with OneShotServer(handler2) as srv:
|
|
client = PrivacyClient(socket_path=srv.sock_path, timeout=2)
|
|
client.call("service.health", {})
|
|
|
|
assert received["id"] != second_id["id"]
|
|
|
|
|
|
def test_call_missing_socket_raises():
|
|
tmp = Path(tempfile.gettempdir()) / f"nonexistent-{uuid.uuid4().hex}.sock"
|
|
client = PrivacyClient(socket_path=tmp, timeout=1)
|
|
with pytest.raises(PrivacyClientError, match="(?i)socket|missing|not found|connect|no such"):
|
|
client.call("service.health", {})
|
|
|
|
|
|
def test_call_timeout_raises():
|
|
def handler(conn):
|
|
# never respond, just sleep longer than client timeout
|
|
time.sleep(3)
|
|
|
|
with OneShotServer(handler) as srv:
|
|
client = PrivacyClient(socket_path=srv.sock_path, timeout=0.3)
|
|
with pytest.raises(PrivacyClientError, match="(?i)timeout|timed out"):
|
|
client.call("service.health", {})
|
|
|
|
|
|
def test_call_malformed_json_response_raises():
|
|
def handler(conn):
|
|
_ = read_one_line(conn)
|
|
conn.sendall(b"not-json\n")
|
|
|
|
with OneShotServer(handler) as srv:
|
|
client = PrivacyClient(socket_path=srv.sock_path, timeout=2)
|
|
with pytest.raises(PrivacyClientError, match="(?i)malformed|invalid|json"):
|
|
client.call("service.health", {})
|
|
|
|
|
|
def test_call_mismatched_id_raises():
|
|
def handler(conn):
|
|
obj = read_one_line(conn)
|
|
resp = {"id": "different-" + obj["id"], "ok": True, "result": {}}
|
|
conn.sendall((json.dumps(resp) + "\n").encode())
|
|
|
|
with OneShotServer(handler) as srv:
|
|
client = PrivacyClient(socket_path=srv.sock_path, timeout=2)
|
|
with pytest.raises(PrivacyClientError, match="(?i)mismatch|id"):
|
|
client.call("service.health", {})
|
|
|
|
|
|
def test_call_ok_false_raises():
|
|
def handler(conn):
|
|
obj = read_one_line(conn)
|
|
resp = {"id": obj["id"], "ok": False, "error": "forbidden"}
|
|
conn.sendall((json.dumps(resp) + "\n").encode())
|
|
|
|
with OneShotServer(handler) as srv:
|
|
client = PrivacyClient(socket_path=srv.sock_path, timeout=2)
|
|
with pytest.raises(PrivacyClientError, match="(?i)forbidden|ok.*false|false"):
|
|
client.call("service.health", {})
|
|
|
|
|
|
def test_call_payload_too_large_before_connect():
|
|
# 64 KiB limit
|
|
large_arg = "x" * (70 * 1024)
|
|
# Use a non-existent socket path; should fail on size check BEFORE attempting connect
|
|
# So we can tell it didn't try to connect if error mentions size
|
|
tmp = Path(tempfile.gettempdir()) / f"nonexistent-{uuid.uuid4().hex}.sock"
|
|
client = PrivacyClient(socket_path=tmp, timeout=1)
|
|
with pytest.raises(PrivacyClientError, match="(?i)64|size|large|payload|KiB"):
|
|
client.call("service.health", {"big": large_arg})
|
|
|
|
# Also test just over limit with real server not needed - ensure no socket file created attempt is made
|
|
# To be sure it didn't connect, we use a server and check that handler was NOT called
|
|
called = {"yes": False}
|
|
|
|
def handler(conn):
|
|
called["yes"] = True
|
|
obj = read_one_line(conn)
|
|
resp = {"id": obj["id"], "ok": True}
|
|
conn.sendall((json.dumps(resp) + "\n").encode())
|
|
|
|
with OneShotServer(handler) as srv:
|
|
client = PrivacyClient(socket_path=srv.sock_path, timeout=2)
|
|
with pytest.raises(PrivacyClientError, match="(?i)size|large|payload|64"):
|
|
client.call("op", {"big": large_arg})
|
|
# give short time for any unwanted connection
|
|
time.sleep(0.2)
|
|
assert not called["yes"], "should not have connected when payload too large"
|
|
|
|
|
|
def test_call_no_arguments_defaults_to_empty():
|
|
def handler(conn):
|
|
obj = read_one_line(conn)
|
|
assert obj["arguments"] == {}
|
|
resp = {"id": obj["id"], "ok": True, "result": "empty-ok"}
|
|
conn.sendall((json.dumps(resp) + "\n").encode())
|
|
|
|
with OneShotServer(handler) as srv:
|
|
client = PrivacyClient(socket_path=srv.sock_path, timeout=2)
|
|
resp = client.call("calendar.list")
|
|
assert resp["result"] == "empty-ok"
|