120 lines
3.9 KiB
Python
120 lines
3.9 KiB
Python
"""Tests for myagents.commands.ollama and myagents.ollama_adapter."""
|
|
|
|
import subprocess
|
|
|
|
import click
|
|
import pytest
|
|
|
|
from myagents.commands import ollama
|
|
from myagents.ollama_adapter import normalize_system
|
|
|
|
|
|
class TestNormalizeSystem:
|
|
def test_moves_system_message_to_top_level(self) -> None:
|
|
body = {
|
|
"system": [{"type": "text", "text": "SYS1"}],
|
|
"messages": [
|
|
{"role": "user", "content": [{"type": "text", "text": "hi"}]},
|
|
{
|
|
"role": "system",
|
|
"content": [{"type": "text", "text": "SYS2"}],
|
|
},
|
|
],
|
|
}
|
|
out = normalize_system(body)
|
|
assert [m["role"] for m in out["messages"]] == ["user"]
|
|
assert [b["text"] for b in out["system"]] == ["SYS1", "SYS2"]
|
|
|
|
def test_keeps_non_system_order(self) -> None:
|
|
body = {
|
|
"messages": [
|
|
{"role": "user", "content": "a"},
|
|
{
|
|
"role": "system",
|
|
"content": [{"type": "text", "text": "s"}],
|
|
},
|
|
{"role": "user", "content": "b"},
|
|
]
|
|
}
|
|
out = normalize_system(body)
|
|
assert [m["role"] for m in out["messages"]] == ["user", "user"]
|
|
assert [m["content"] for m in out["messages"]] == ["a", "b"]
|
|
|
|
def test_string_system_untouched(self) -> None:
|
|
body = {"system": "top", "messages": [{"role": "user", "content": "a"}]}
|
|
out = normalize_system(body)
|
|
assert out["system"] == [{"type": "text", "text": "top"}]
|
|
|
|
|
|
class TestSystemdUnit:
|
|
def test_unit_text_smoke(self) -> None:
|
|
text = ollama._sysd_unit_text()
|
|
assert "ExecStart=" in text
|
|
assert "-m myagents.ollama_adapter" in text
|
|
assert "WantedBy=default.target" in text
|
|
|
|
def test_sysd_start_enables_and_reloads(
|
|
self, monkeypatch, tmp_path
|
|
) -> None:
|
|
calls: list[list[str]] = []
|
|
|
|
def fake_run(cmd, **kwargs):
|
|
kwargs.pop("capture_output", None)
|
|
kwargs.pop("text", None)
|
|
calls.append(list(cmd))
|
|
return subprocess.CompletedProcess(cmd, 0)
|
|
|
|
monkeypatch.setattr(ollama.subprocess, "run", fake_run)
|
|
monkeypatch.setattr(ollama, "UNIT_DIR", tmp_path)
|
|
monkeypatch.setattr(ollama, "UNIT_PATH", tmp_path / ollama.UNIT_NAME)
|
|
|
|
ollama._sysd_start()
|
|
|
|
assert ["systemctl", "--user", "daemon-reload"] in calls
|
|
assert [
|
|
"systemctl",
|
|
"--user",
|
|
"enable",
|
|
"--now",
|
|
ollama.UNIT_NAME,
|
|
] in calls
|
|
assert any(c[0] == "loginctl" and "enable-linger" in c for c in calls)
|
|
assert (tmp_path / ollama.UNIT_NAME).exists()
|
|
|
|
def test_sysd_stop_disables(self, monkeypatch) -> None:
|
|
calls: list[list[str]] = []
|
|
|
|
def fake_run(cmd, **kwargs):
|
|
kwargs.pop("capture_output", None)
|
|
kwargs.pop("text", None)
|
|
calls.append(list(cmd))
|
|
return subprocess.CompletedProcess(cmd, 0)
|
|
|
|
monkeypatch.setattr(ollama.subprocess, "run", fake_run)
|
|
ollama._sysd_stop()
|
|
|
|
assert [
|
|
"systemctl",
|
|
"--user",
|
|
"disable",
|
|
"--now",
|
|
ollama.UNIT_NAME,
|
|
] in calls
|
|
assert ["systemctl", "--user", "daemon-reload"] in calls
|
|
|
|
|
|
class TestBackendDispatch:
|
|
def test_darwin_launchd(self) -> None:
|
|
# Test runs on macOS; _platform() returns the real platform.
|
|
if ollama._platform() == "darwin":
|
|
assert ollama._backend().name == "launchd"
|
|
|
|
def test_linux_systemd(self, monkeypatch) -> None:
|
|
monkeypatch.setattr(ollama, "_platform", lambda: "linux")
|
|
assert ollama._backend().name == "systemd"
|
|
|
|
def test_unsupported_platform_raises(self, monkeypatch) -> None:
|
|
monkeypatch.setattr(ollama, "_platform", lambda: "win32")
|
|
with pytest.raises(click.ClickException):
|
|
ollama._backend()
|