my-project-public

#1

import os

#2

import pytest

#3

from unittest.mock import patch, MagicMock

#4

#5

from mnemosyne.core import local_llm

#6

from mnemosyne.core.llm_backends import (

#7

CallableLLMBackend,

#8

set_host_llm_backend,

)

class TestRemoteLLM:

def test_llm_available_returns_true_when_base_url_set(self, monkeypatch):

#14

"""BUG-2: llm_available() must report True when remote is configured."""

#15

monkeypatch.setenv("MNEMOSYNE_LLM_BASE_URL", "http://localhost:8080/v1")

#16

# Reset module-level cache

#17

monkeypatch.setattr(local_llm, "LLM_BASE_URL", "http://localhost:8080/v1")

#18

monkeypatch.setattr(local_llm, "_llm_available", None)

#19

monkeypatch.setattr(local_llm, "_llm_instance", None)

#20

#21

assert local_llm.llm_available() is True

#22

#23

def test_call_remote_llm_with_mock_response(self, monkeypatch):

#24

"""BUG-2: _call_remote_llm parses OpenAI-compatible response correctly."""

#25

monkeypatch.setenv("MNEMOSYNE_LLM_BASE_URL", "http://test-server/v1")

#26

monkeypatch.setenv("MNEMOSYNE_LLM_API_KEY", "sk-test")

#27

monkeypatch.setattr(local_llm, "LLM_BASE_URL", "http://test-server/v1")

#28

monkeypatch.setattr(local_llm, "LLM_API_KEY", "sk-test")

#29

monkeypatch.setattr(local_llm, "LLM_REMOTE_MODEL", "test-model")

#30

monkeypatch.setattr(local_llm, "LLM_MAX_TOKENS", 128)

mock_response = {

"choices": [

{"message": {"content": "This is a test summary."}}

]

}

# Mock httpx by patching the import inside _call_remote_llm

#39

mock_client = MagicMock()

#40

mock_client.post.return_value.raise_for_status = lambda: None

#41

mock_client.post.return_value.json.return_value = mock_response

#42

mock_client.__enter__ = lambda s: s

#43

mock_client.__exit__ = lambda *args: None

#44

#45

mock_httpx_module = MagicMock()

#46

mock_httpx_module.Client = MagicMock(return_value=mock_client)

#47

#48

# Save original import to avoid recursion

#49

_orig_import = __builtins__["__import__"] if isinstance(__builtins__, dict) else builtins.__import__

#50

def mock_import(name, *args, **kwargs):

#51

if name == "httpx":

#52

return mock_httpx_module

#53

return _orig_import(name, *args, **kwargs)

#54

#55

with patch("builtins.__import__", mock_import):

#56

result = local_llm._call_remote_llm("Test prompt")

#57

assert result == "This is a test summary."

#58

#59

# Verify the call was made with correct payload

#60

call_args = mock_client.post.call_args

#61

assert call_args[0][0] == "http://test-server/v1/chat/completions"

#62

payload = call_args[1]["json"]

#63

assert payload["model"] == "test-model"

#64

assert payload["messages"][0]["content"] == "Test prompt"

#65

assert "Authorization" in call_args[1]["headers"]

#66

#67

def test_call_remote_llm_urllib_fallback(self, monkeypatch):

#68

"""BUG-2: Falls back to urllib when httpx unavailable."""

#69

monkeypatch.setenv("MNEMOSYNE_LLM_BASE_URL", "http://test-server/v1")

#70

monkeypatch.setattr(local_llm, "LLM_BASE_URL", "http://test-server/v1")

#71

monkeypatch.setattr(local_llm, "LLM_API_KEY", "")

#72

monkeypatch.setattr(local_llm, "LLM_MAX_TOKENS", 128)

mock_response = {

"choices": [

{"message": {"content": "Fallback summary."}}

]

}

import json

mock_data = json.dumps(mock_response).encode()

class MockResponse:

def read(self):

return mock_data

def __enter__(self):

return self

def __exit__(self, *args):

#89

pass

#90

#91

# Patch httpx import in local_llm module to simulate it not being installed

#92

with patch.dict("sys.modules", {"httpx": None}):

#93

with patch("urllib.request.urlopen", return_value=MockResponse()):

#94

result = local_llm._call_remote_llm("Test prompt")

#95

assert result == "Fallback summary."

#96

#97

def test_summarize_memories_prefers_remote_over_local(self, monkeypatch):

#98

"""BUG-2: summarize_memories() calls remote when BASE_URL is set."""

#99

monkeypatch.setenv("MNEMOSYNE_LLM_BASE_URL", "http://remote/v1")

#100

monkeypatch.setattr(local_llm, "LLM_BASE_URL", "http://remote/v1")

#101

monkeypatch.setattr(local_llm, "_llm_available", False)

#102

monkeypatch.setattr(local_llm, "_llm_instance", None)

#103

#104

with patch.object(local_llm, "_call_remote_llm", return_value="Remote summary.") as mock_remote:

#105

result = local_llm.summarize_memories(["Memory one", "Memory two"])

#106

assert result == "Remote summary."

#107

mock_remote.assert_called_once()

#108

#109

def test_summarize_memories_falls_back_local_when_remote_fails(self, monkeypatch):

#110

"""BUG-2: When remote fails and local is unavailable, return None (aaak fallback)."""

#111

monkeypatch.setenv("MNEMOSYNE_LLM_BASE_URL", "http://remote/v1")

#112

monkeypatch.setattr(local_llm, "LLM_BASE_URL", "http://remote/v1")

#113

#114

# Remote returns None (failure), local _load_llm returns None (unavailable)

#115

with patch.object(local_llm, "_call_remote_llm", return_value=None) as mock_remote:

#116

with patch.object(local_llm, "_load_llm", return_value=None) as mock_load:

#117

result = local_llm.summarize_memories(["Memory one"])

#118

# Should return None since both remote and local fail

#119

assert result is None

#120

mock_remote.assert_called_once()

#121

mock_load.assert_called_once()

#122

#123

#124

class TestHostLLMBackend:

#125

"""Tests for the host LLM adapter integration in summarize_memories()."""

#126

#127

def _enable_host(self, monkeypatch):

#128

monkeypatch.setattr(local_llm, "LLM_ENABLED", True)

#129

monkeypatch.setattr(local_llm, "HOST_LLM_ENABLED", True)

#130

monkeypatch.setattr(local_llm, "HOST_LLM_PROVIDER", None)

#131

monkeypatch.setattr(local_llm, "HOST_LLM_MODEL", None)

#132

#133

def test_summarize_memories_uses_host_when_enabled(self, monkeypatch):

#134

"""Host backend is consulted before remote when enabled."""

#135

self._enable_host(monkeypatch)

#136

monkeypatch.setattr(local_llm, "LLM_BASE_URL", "http://remote/v1")

#137

monkeypatch.setattr(local_llm, "LLM_MAX_TOKENS", 128)

#138

monkeypatch.setattr(local_llm, "HOST_LLM_PROVIDER", "openai-codex")

#139

monkeypatch.setattr(local_llm, "HOST_LLM_MODEL", "gpt-5.1-mini")

captured = []

def fake(prompt, *, max_tokens, temperature, timeout, provider=None, model=None):

#144

captured.append({

#145

"prompt": prompt,

#146

"max_tokens": max_tokens,

#147

"temperature": temperature,

#148

"timeout": timeout,

#149

"provider": provider,

#150

"model": model,

#151

})

#152

return "Host summary."

#153

#154

set_host_llm_backend(CallableLLMBackend("test", fake))

#155

with patch.object(local_llm, "_call_remote_llm") as mock_remote, \

#156

patch.object(local_llm, "_call_local_llm") as mock_local:

#157

assert local_llm.summarize_memories(["Memory one"]) == "Host summary."

#158

mock_remote.assert_not_called()

#159

mock_local.assert_not_called()

#160

assert captured

#161

assert captured[0]["max_tokens"] == 128

#162

assert captured[0]["temperature"] == 0.3

#163

assert captured[0]["timeout"] == local_llm.HOST_LLM_TIMEOUT

#164

assert captured[0]["provider"] == "openai-codex"

#165

assert captured[0]["model"] == "gpt-5.1-mini"

#166

# Host prompt MUST NOT contain TinyLlama chat-template tokens.

#167

assert "<|user|>" not in captured[0]["prompt"]

#168

assert "</s>" not in captured[0]["prompt"]

#169

assert "<|assistant|>" not in captured[0]["prompt"]

#170

#171

def test_summarize_memories_skips_remote_on_host_miss(self, monkeypatch):

#172

"""A3 contract: host enabled + host returns None → fall to local, NOT to remote."""

#173

self._enable_host(monkeypatch)

#174

monkeypatch.setattr(local_llm, "LLM_BASE_URL", "http://remote/v1")

#175

set_host_llm_backend(CallableLLMBackend("test", lambda *a, **k: None))

#176

with patch.object(local_llm, "_call_remote_llm", return_value="Remote summary.") as mock_remote, \

#177

patch.object(local_llm, "_call_local_llm", return_value="Local summary.") as mock_local:

#178

assert local_llm.summarize_memories(["Memory one"]) == "Local summary."

#179

mock_remote.assert_not_called()

#180

mock_local.assert_called_once()

#181

#182

def test_summarize_memories_returns_none_when_host_and_local_both_fail(self, monkeypatch):

#183

"""Host attempted + nothing + local fails → None (NOT remote)."""

#184

self._enable_host(monkeypatch)

#185

monkeypatch.setattr(local_llm, "LLM_BASE_URL", "http://remote/v1")

#186

set_host_llm_backend(CallableLLMBackend("test", lambda *a, **k: None))

#187

with patch.object(local_llm, "_call_remote_llm", return_value="Remote summary.") as mock_remote, \

#188

patch.object(local_llm, "_call_local_llm", return_value=None) as mock_local:

#189

assert local_llm.summarize_memories(["Memory one"]) is None

#190

mock_remote.assert_not_called()

#191

mock_local.assert_called_once()

#192

#193

def test_summarize_memories_unchanged_when_HOST_LLM_ENABLED_false(self, monkeypatch):

#194

"""REGRESSION: existing remote/local behavior is preserved when host is off."""

#195

monkeypatch.setattr(local_llm, "LLM_ENABLED", True)

#196

monkeypatch.setattr(local_llm, "HOST_LLM_ENABLED", False) # explicitly off

#197

monkeypatch.setattr(local_llm, "LLM_BASE_URL", "http://remote/v1")

#198

# Even with a backend registered, host is gated off.

#199

set_host_llm_backend(CallableLLMBackend("test", lambda *a, **k: "Host summary."))

#200

with patch.object(local_llm, "_call_remote_llm", return_value="Remote summary.") as mock_remote:

#201

assert local_llm.summarize_memories(["Memory one"]) == "Remote summary."

#202

mock_remote.assert_called_once()

#203

#204

def test_summarize_memories_unchanged_when_LLM_ENABLED_false(self, monkeypatch):

#205

"""A2 contract: MNEMOSYNE_LLM_ENABLED=false disables host and remote alike."""

#206

monkeypatch.setattr(local_llm, "LLM_ENABLED", False)

#207

monkeypatch.setattr(local_llm, "HOST_LLM_ENABLED", True)

#208

monkeypatch.setattr(local_llm, "LLM_BASE_URL", "http://remote/v1")

#209

set_host_llm_backend(CallableLLMBackend("test", lambda *a, **k: "Host summary."))

#210

with patch.object(local_llm, "_call_remote_llm", return_value="Remote summary.") as mock_remote, \

#211

patch.object(local_llm, "_call_local_llm", return_value=None) as mock_local:

#212

# Host gated by LLM_ENABLED → not attempted; remote also gated → not called;

#213

# local: _call_local_llm internally checks via _load_llm() which itself

#214

# gates on LLM_ENABLED (preserving prior behavior). End result: None.

#215

assert local_llm.summarize_memories(["Memory one"]) is None

#216

mock_remote.assert_not_called()

#217

#218

def test_summarize_memories_swallows_host_exception(self, monkeypatch):

#219

"""Backend that raises is treated as host-attempted-with-no-output (A3 still applies)."""

#220

self._enable_host(monkeypatch)

#221

monkeypatch.setattr(local_llm, "LLM_BASE_URL", "http://remote/v1")

#222

#223

def boom(*a, **k):

#224

raise RuntimeError("provider exploded")

#225

#226

set_host_llm_backend(CallableLLMBackend("test", boom))

#227

with patch.object(local_llm, "_call_remote_llm", return_value="Remote summary.") as mock_remote, \

#228

patch.object(local_llm, "_call_local_llm", return_value="Local summary.") as mock_local:

#229

assert local_llm.summarize_memories(["Memory one"]) == "Local summary."

#230

mock_remote.assert_not_called()

#231

mock_local.assert_called_once()

#232

#233

#234

class TestLLMAvailable:

#235

"""Tests for the host-aware llm_available() gate."""

#236

#237

def test_llm_available_true_when_only_host_backend_registered(self, monkeypatch):

#238

"""A5 contract: Hermes-only users (no remote URL, no GGUF) still report available."""

#239

monkeypatch.setattr(local_llm, "LLM_ENABLED", True)

#240

monkeypatch.setattr(local_llm, "HOST_LLM_ENABLED", True)

#241

monkeypatch.setattr(local_llm, "LLM_BASE_URL", "")

#242

monkeypatch.setattr(local_llm, "_llm_available", False)

#243

set_host_llm_backend(CallableLLMBackend("test", lambda *a, **k: "x"))

#244

assert local_llm.llm_available() is True

#245

#246

def test_llm_available_false_when_host_enabled_but_no_backend(self, monkeypatch):

#247

"""HOST_LLM_ENABLED=true with no backend registered must not fake availability."""

#248

monkeypatch.setattr(local_llm, "LLM_ENABLED", True)

#249

monkeypatch.setattr(local_llm, "HOST_LLM_ENABLED", True)

#250

monkeypatch.setattr(local_llm, "LLM_BASE_URL", "")

#251

monkeypatch.setattr(local_llm, "_llm_available", False)

#252

# No backend registered.

#253

assert local_llm.llm_available() is False

#254

#255

def test_llm_available_false_when_LLM_ENABLED_false(self, monkeypatch):

#256

"""A2 contract: MNEMOSYNE_LLM_ENABLED=false makes everything unavailable."""

#257

monkeypatch.setattr(local_llm, "LLM_ENABLED", False)

#258

monkeypatch.setattr(local_llm, "HOST_LLM_ENABLED", True)

#259

monkeypatch.setattr(local_llm, "LLM_BASE_URL", "http://remote/v1")

#260

monkeypatch.setattr(local_llm, "_llm_available", False)

#261

set_host_llm_backend(CallableLLMBackend("test", lambda *a, **k: "x"))

#262

assert local_llm.llm_available() is False

#263

#264

#265

class TestHostAwareChunking:

#266

"""Tests for HOST_LLM_N_CTX-aware budgeting (decision C6)."""

#267

#268

def test_prompt_token_budget_uses_host_n_ctx_when_host_will_handle(self, monkeypatch):

#269

monkeypatch.setattr(local_llm, "LLM_ENABLED", True)

#270

monkeypatch.setattr(local_llm, "HOST_LLM_ENABLED", True)

#271

monkeypatch.setattr(local_llm, "LLM_N_CTX", 2048)

#272

monkeypatch.setattr(local_llm, "HOST_LLM_N_CTX", 32000)

#273

monkeypatch.setattr(local_llm, "LLM_MAX_TOKENS", 256)

#274

set_host_llm_backend(CallableLLMBackend("test", lambda *a, **k: "x"))

#275

#276

host_budget = local_llm._prompt_token_budget()

#277

# Should be much larger than the TinyLlama-calibrated default budget.

#278

assert host_budget > 10_000

#279

#280

# Same module without a host backend → falls back to LLM_N_CTX budget.

#281

set_host_llm_backend(None)

#282

local_budget = local_llm._prompt_token_budget()

#283

assert local_budget < host_budget

#284

z6Mkq5mY3JWtxoxUobWcfNHm7AkRubgSWEZTkBVqZXJviFZ5/my-project-public