feat(ollama): Anthropic→Ollama 归一化 adapter + launchd/systemd 常驻管理

tests/test_ollama.py

@@ -0,0 +1,119 @@
+"""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()