Adolfo Reyna
75 lines
2.6 KiB
Python
75 lines
2.6 KiB
Python
|
|
import pytest
|
|
from unittest.mock import MagicMock, patch
|
|
from reyna_cli.apple_llm_client import AppleLLMClient, get_apple_llm_session, apple_llm_close, apple_llm_status
|
|
|
|
@pytest.fixture
|
|
def mock_apple_llm():
|
|
with patch("subprocess.Popen") as mock_popen, \
|
|
patch("subprocess.run") as mock_run, \
|
|
patch("pathlib.Path.write_text"), \
|
|
patch("shutil.rmtree"):
|
|
|
|
# Mock the binary build
|
|
mock_run.return_value = MagicMock(returncode=0)
|
|
|
|
# Mock the process
|
|
mock_proc = MagicMock()
|
|
mock_proc.poll.return_value = None
|
|
mock_proc.pid = 1234
|
|
mock_proc.stdout.readline.side_effect = [
|
|
'{"id": "0", "ok": true, "text": "Polished text", "ms": 100, "mode": "line"}\\n',
|
|
'{"id": "1", "ok": true, "text": "Quick reply", "ms": 50, "mode": "quick_reply"}\\n',
|
|
'{"id": "2", "ok": true, "text": "Chat response", "ms": 200, "mode": "chat"}\\n',
|
|
'', # EOF
|
|
]
|
|
mock_popen.return_value = mock_proc
|
|
|
|
client = AppleLLMClient()
|
|
yield client
|
|
|
|
def test_apple_llm_lifecycle(mock_apple_llm):
|
|
# Test status before start
|
|
status = apple_llm_status()
|
|
assert status["active"] is False
|
|
|
|
# Start and check status
|
|
mock_apple_llm.start()
|
|
status = apple_llm_status()
|
|
# Note: apple_llm_status uses the global singleton, not the fixture instance.
|
|
# For the purpose of these tests, we'll mock the global session if needed,
|
|
# but let's focus on the Client logic first.
|
|
|
|
def test_apple_llm_call_success(mock_apple_llm):
|
|
payload = {"mode": "line", "text": "Hello world"}
|
|
mock_proc = MagicMock()
|
|
mock_apple_llm.proc = mock_proc
|
|
|
|
def reply_on_write(_: str) -> None:
|
|
event, result_box = mock_apple_llm.pending["0"]
|
|
result_box["data"] = {"ok": True, "text": "Polished!"}
|
|
event.set()
|
|
|
|
mock_proc.stdin.write.side_effect = reply_on_write
|
|
with patch.object(mock_apple_llm, "start"):
|
|
res = mock_apple_llm.call(payload)
|
|
|
|
assert res["ok"] is True
|
|
assert res["text"] == "Polished!"
|
|
|
|
def test_apple_llm_timeout(mock_apple_llm):
|
|
with patch("threading.Event.wait", return_value=False):
|
|
res = mock_apple_llm.call({"mode": "line", "text": "test"}, timeout=0.1)
|
|
assert res["ok"] is False
|
|
assert "timeout" in res["error"]
|
|
|
|
def test_apple_llm_check_mock(mock_apple_llm):
|
|
with patch("subprocess.run") as mock_run:
|
|
mock_run.return_value = MagicMock(
|
|
stdout='{"ok": true, "available": true, "ping": "ok"}',
|
|
returncode=0
|
|
)
|
|
res = mock_apple_llm.check()
|
|
assert res["ok"] is True
|
|
assert res["available"] is True
|