anti
f2b3393669
Replaces LICENSE (GPLv3 -> AGPLv3) and prepends `SPDX-License-Identifier: AGPL-3.0-or-later` to every source file across decnet/, decnet_web/, tests/, scripts/, and tools/. Rationale: closes the GPLv3 ASP loophole so any party operating a modified DECNET as a network service must offer their modified source. Personal copyright (Samuel Paschuan) + inbound=outbound contributions make a future unilateral relicense infeasible. - LICENSE: full AGPL-3.0 text (gnu.org/licenses/agpl-3.0.txt) - COPYRIGHT: project copyright notice - tools/add_spdx_headers.py: idempotent header injector (shebang- and PEP 263-aware) Touches 1565 source files (.py, .ts, .tsx, .js, .jsx, .css, .sh). No behavior change; comments only.
142 lines
4.6 KiB
Python
142 lines
4.6 KiB
Python
# SPDX-License-Identifier: AGPL-3.0-or-later
|
|
"""LLM backend factory + Ollama implementation."""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
|
|
import pytest
|
|
|
|
from decnet.realism.llm import LLMTimeout, get_llm
|
|
from decnet.realism.llm.impl.fake import FakeBackend
|
|
from decnet.realism.llm.impl.ollama import OllamaBackend
|
|
|
|
|
|
# ── factory dispatch ─────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_factory_default_is_ollama(monkeypatch):
|
|
monkeypatch.delenv("DECNET_REALISM_LLM", raising=False)
|
|
monkeypatch.setattr("decnet.realism.llm.config._cached_backend", None)
|
|
backend = get_llm()
|
|
assert isinstance(backend, OllamaBackend)
|
|
|
|
|
|
def test_factory_selects_fake(monkeypatch):
|
|
monkeypatch.setenv("DECNET_REALISM_LLM", "fake")
|
|
monkeypatch.setattr("decnet.realism.llm.config._cached_backend", None)
|
|
backend = get_llm()
|
|
assert isinstance(backend, FakeBackend)
|
|
|
|
|
|
def test_factory_unknown_raises(monkeypatch):
|
|
monkeypatch.setenv("DECNET_REALISM_LLM", "vllm-someday")
|
|
monkeypatch.setattr("decnet.realism.llm.config._cached_backend", None)
|
|
with pytest.raises(ValueError, match="Unsupported"):
|
|
get_llm()
|
|
|
|
|
|
def test_factory_passes_model_through(monkeypatch):
|
|
monkeypatch.setenv("DECNET_REALISM_LLM", "ollama")
|
|
backend = get_llm(model="qwen2:7b")
|
|
assert backend.model == "qwen2:7b"
|
|
|
|
|
|
# ── FakeBackend ──────────────────────────────────────────────────────────────
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fake_backend_returns_canned_output():
|
|
fb = FakeBackend(output="Subject: hi\n\nbody")
|
|
result = await fb.generate("any prompt")
|
|
assert result.success is True
|
|
assert result.text.startswith("Subject:")
|
|
assert result.model == "fake-model"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fake_backend_can_simulate_failure():
|
|
fb = FakeBackend(success=False)
|
|
result = await fb.generate("prompt")
|
|
assert result.success is False
|
|
assert result.text == ""
|
|
|
|
|
|
# ── OllamaBackend (subprocess stubbed) ───────────────────────────────────────
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ollama_backend_success(monkeypatch):
|
|
"""Stub asyncio.create_subprocess_exec to return canned stdout."""
|
|
|
|
class _StubProc:
|
|
returncode = 0
|
|
|
|
async def communicate(self, _stdin):
|
|
return b"Subject: hi\n\nbody\n", b""
|
|
|
|
async def fake_create(*args, **kwargs): # noqa: ARG001
|
|
return _StubProc()
|
|
|
|
monkeypatch.setattr(asyncio, "create_subprocess_exec", fake_create)
|
|
|
|
backend = OllamaBackend(model="m1", timeout=1.0)
|
|
result = await backend.generate("hello")
|
|
assert result.success is True
|
|
assert "Subject:" in result.text
|
|
assert result.model == "m1"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ollama_backend_non_zero_rc_marks_failure(monkeypatch):
|
|
class _StubProc:
|
|
returncode = 1
|
|
|
|
async def communicate(self, _stdin):
|
|
return b"", b"model not found"
|
|
|
|
async def fake_create(*args, **kwargs): # noqa: ARG001
|
|
return _StubProc()
|
|
|
|
monkeypatch.setattr(asyncio, "create_subprocess_exec", fake_create)
|
|
|
|
backend = OllamaBackend(model="m1", timeout=1.0)
|
|
result = await backend.generate("hello")
|
|
assert result.success is False
|
|
assert result.extra["rc"] == 1
|
|
assert "model not found" in result.extra["stderr"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ollama_backend_timeout_raises(monkeypatch):
|
|
class _StubProc:
|
|
returncode = None
|
|
|
|
async def communicate(self, _stdin):
|
|
await asyncio.sleep(10) # well past the timeout
|
|
return b"", b""
|
|
|
|
def kill(self):
|
|
pass
|
|
|
|
async def fake_create(*args, **kwargs): # noqa: ARG001
|
|
return _StubProc()
|
|
|
|
monkeypatch.setattr(asyncio, "create_subprocess_exec", fake_create)
|
|
|
|
backend = OllamaBackend(model="m1", timeout=0.05)
|
|
with pytest.raises(LLMTimeout):
|
|
await backend.generate("hello")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ollama_backend_missing_binary_returns_failure(monkeypatch):
|
|
async def fake_create(*args, **kwargs): # noqa: ARG001
|
|
raise FileNotFoundError("ollama: not found")
|
|
|
|
monkeypatch.setattr(asyncio, "create_subprocess_exec", fake_create)
|
|
|
|
backend = OllamaBackend(model="m1", timeout=1.0)
|
|
result = await backend.generate("hello")
|
|
assert result.success is False
|
|
assert result.extra["rc"] == 127
|