solpbc.org
personal memory agent
1# SPDX-License-Identifier: AGPL-3.0-only
2# Copyright (c) 2026 sol pbc
3
4import importlib
5import sys
6import types
7from unittest.mock import Mock
8
9import numpy as np
10import pytest
11
12from think.utils import now_ms
13
14
15@pytest.fixture(autouse=True)
16def set_test_journal_path(request, monkeypatch):
17 """Set _SOLSTONE_JOURNAL_OVERRIDE to tests/fixtures/journal for all unit tests.
18
19 This ensures all tests have a valid _SOLSTONE_JOURNAL_OVERRIDE without needing
20 to explicitly set it in each test. Integration tests are excluded.
21 """
22 # Skip for integration tests - they may have different requirements
23 if "integration" in request.node.keywords:
24 return
25
26 # Set _SOLSTONE_JOURNAL_OVERRIDE to tests/fixtures/journal for all unit tests
27 monkeypatch.setenv("_SOLSTONE_JOURNAL_OVERRIDE", "tests/fixtures/journal")
28
29
30@pytest.fixture(autouse=True)
31def add_module_stubs(request, monkeypatch):
32 # Skip stubbing for integration tests
33 if "integration" in request.node.keywords:
34 return
35
36 # stub heavy modules used by think.indexer
37 if "usearch.index" not in sys.modules:
38 usearch = types.ModuleType("usearch")
39 index_mod = types.ModuleType("usearch.index")
40
41 class DummyIndex:
42 def __init__(self, *a, **k):
43 pass
44
45 def save(self, *a, **k):
46 pass
47
48 @classmethod
49 def restore(cls, *a, **k):
50 return cls()
51
52 def remove(self, *a, **k):
53 pass
54
55 def add(self, *a, **k):
56 pass
57
58 def search(self, *a, **k):
59 class Res:
60 keys = [1]
61 distances = [0.0]
62
63 return Res()
64
65 index_mod.Index = DummyIndex
66 usearch.index = index_mod
67 sys.modules["usearch"] = usearch
68 sys.modules["usearch.index"] = index_mod
69 if "sentence_transformers" not in sys.modules:
70 st_mod = types.ModuleType("sentence_transformers")
71
72 class DummyST:
73 def __init__(self, *a, **k):
74 pass
75
76 def get_sentence_embedding_dimension(self):
77 return 384
78
79 def encode(self, texts):
80 if isinstance(texts, str):
81 texts = [texts]
82 return [([0.0] * 384) for _ in texts]
83
84 st_mod.SentenceTransformer = DummyST
85 sys.modules["sentence_transformers"] = st_mod
86 if "sklearn.metrics.pairwise" not in sys.modules:
87 pairwise = types.ModuleType("pairwise")
88
89 def cosine_similarity(a, b):
90 return [[1.0]]
91
92 pairwise.cosine_similarity = cosine_similarity
93 metrics = types.ModuleType("metrics")
94 metrics.pairwise = pairwise
95
96 cluster = types.ModuleType("sklearn.cluster")
97
98 class DummyHDBSCAN:
99 def __init__(self, **k):
100 pass
101
102 def fit(self, X):
103 self.labels_ = np.full(len(X), -1, dtype=int)
104 return self
105
106 cluster.HDBSCAN = DummyHDBSCAN
107
108 sklearn = types.ModuleType("sklearn")
109 sklearn.metrics = metrics
110 sklearn.cluster = cluster
111 sys.modules["sklearn"] = sklearn
112 sys.modules["sklearn.metrics"] = metrics
113 sys.modules["sklearn.metrics.pairwise"] = pairwise
114 sys.modules["sklearn.cluster"] = cluster
115 if "dotenv" not in sys.modules:
116 dotenv_mod = types.ModuleType("dotenv")
117
118 def load_dotenv(*a, **k):
119 return True
120
121 def dotenv_values(*a, **k):
122 return {}
123
124 dotenv_mod.load_dotenv = load_dotenv
125 dotenv_mod.dotenv_values = dotenv_values
126 sys.modules["dotenv"] = dotenv_mod
127 # Import real observe package first to avoid shadowing with stubs
128 if "observe" not in sys.modules:
129 importlib.import_module("observe")
130 if "observe.detect" not in sys.modules:
131 detect_mod = types.ModuleType("observe.detect")
132
133 def input_detect():
134 return None, None
135
136 detect_mod.input_detect = input_detect
137 sys.modules["observe.detect"] = detect_mod
138 observe_pkg = sys.modules.get("observe")
139 setattr(observe_pkg, "detect", detect_mod)
140 if "observe.hear" not in sys.modules:
141 # Import the real module for format_audio and load_transcript
142 hear_mod = importlib.import_module("observe.hear")
143 sys.modules["observe.hear"] = hear_mod
144 observe_pkg = sys.modules.get("observe")
145 setattr(observe_pkg, "hear", hear_mod)
146 if "observe.sense" not in sys.modules:
147 # Import the real module - it has minimal dependencies
148 sense_mod = importlib.import_module("observe.sense")
149 sys.modules["observe.sense"] = sense_mod
150 observe_pkg = sys.modules.get("observe")
151 setattr(observe_pkg, "sense", sense_mod)
152 if "observe.utils" not in sys.modules:
153 # Import the real module
154 utils_mod = importlib.import_module("observe.utils")
155 sys.modules["observe.utils"] = utils_mod
156 observe_pkg = sys.modules.get("observe")
157 setattr(observe_pkg, "utils", utils_mod)
158 if "observe.screen" not in sys.modules:
159 # Import the real module for format_screen
160 screen_mod = importlib.import_module("observe.screen")
161 sys.modules["observe.screen"] = screen_mod
162 observe_pkg = sys.modules.get("observe")
163 setattr(observe_pkg, "screen", screen_mod)
164 if "gi" not in sys.modules:
165 gi_mod = types.ModuleType("gi")
166 gi_mod.require_version = lambda *a, **k: None
167
168 class Dummy(types.ModuleType):
169 pass
170
171 repo = types.ModuleType("gi.repository")
172 repo.Gdk = Dummy("Gdk")
173 repo.Gtk = Dummy("Gtk")
174 gi_mod.repository = repo
175 sys.modules["gi"] = gi_mod
176 sys.modules["gi.repository"] = repo
177 sys.modules["Gdk"] = repo.Gdk
178 sys.modules["Gtk"] = repo.Gtk
179 google_mod = sys.modules.get("google", types.ModuleType("google"))
180 genai_mod = types.ModuleType("google.genai")
181
182 class DummyModels:
183 def generate_content(self, *, model, contents, config=None):
184 return types.SimpleNamespace(text="[]", candidates=[], usage_metadata=None)
185
186 class DummyClient:
187 def __init__(self, *a, **k):
188 self.models = DummyModels()
189
190 genai_mod.Client = DummyClient
191
192 # Mock Content type for type hints
193 class MockContent:
194 pass
195
196 # Mock config builders
197 class MockHttpOptions:
198 def __init__(self, **k):
199 self.timeout = k.get("timeout")
200
201 class MockThinkingConfig:
202 def __init__(self, **k):
203 self.thinking_budget = k.get("thinking_budget")
204
205 class MockGenerateContentConfig:
206 def __init__(self, **k):
207 for key, value in k.items():
208 setattr(self, key, value)
209
210 class MockHttpRetryOptions:
211 def __init__(self, **k):
212 pass
213
214 genai_mod.types = types.SimpleNamespace(
215 GenerateContentConfig=MockGenerateContentConfig,
216 Content=MockContent,
217 HttpOptions=MockHttpOptions,
218 HttpRetryOptions=MockHttpRetryOptions,
219 ThinkingConfig=MockThinkingConfig,
220 )
221 google_mod.genai = genai_mod
222 sys.modules["google"] = google_mod
223 sys.modules["google.genai"] = genai_mod
224 if "cv2" not in sys.modules:
225 cv2_mod = types.ModuleType("cv2")
226 cv2_mod.COLOR_RGB2LAB = 0
227
228 def cvtColor(arr, code):
229 arr = np.asarray(arr)
230 gray = arr.mean(axis=2)
231 return np.stack([gray, gray, gray], axis=2)
232
233 cv2_mod.cvtColor = cvtColor
234 sys.modules["cv2"] = cv2_mod
235 if "soundfile" not in sys.modules:
236 sf_mod = types.ModuleType("soundfile")
237
238 def write(buf, data, samplerate, format=None):
239 buf.write(b"fLaCfake")
240
241 sf_mod.write = write
242 sys.modules["soundfile"] = sf_mod
243 for name in [
244 "noisereduce",
245 ]:
246 if name not in sys.modules:
247 sys.modules[name] = types.ModuleType(name)
248
249
250@pytest.fixture(autouse=True)
251def reset_supervisor_state():
252 """Reset supervisor module state before/after tests to prevent cross-test pollution."""
253 try:
254 import think.supervisor as mod
255
256 # Reset before test
257 mod._daily_state["last_day"] = None
258 mod._is_remote_mode = False
259 # Create fresh task queue
260 mod._task_queue = mod.TaskQueue(on_queue_change=None)
261 except ImportError:
262 pass # supervisor not loaded yet
263 yield
264 try:
265 import think.supervisor as mod
266
267 # Reset after test
268 mod._daily_state["last_day"] = None
269 mod._is_remote_mode = False
270 mod._observer_health = {}
271 mod._enabled_observers = set()
272 # Create fresh task queue
273 mod._task_queue = mod.TaskQueue(on_queue_change=None)
274 except ImportError:
275 pass
276
277
278@pytest.fixture
279def mock_callosum(monkeypatch):
280 """Mock Callosum connections to capture emitted events without real I/O.
281
282 This fixture provides a MockCallosumConnection class that:
283 - Enforces the start-before-emit requirement
284 - Broadcasts events to all listeners (like the real Callosum)
285 - Works without real socket connections
286
287 Usage:
288 def test_example(mock_callosum):
289 from think.callosum import CallosumConnection
290
291 received = []
292 listener = CallosumConnection()
293 listener.start(callback=lambda msg: received.append(msg))
294
295 # Now emit events and they'll be captured in received
296 """
297 all_listeners = []
298
299 class MockCallosumConnection:
300 def __init__(self, socket_path=None):
301 self.socket_path = socket_path
302 self.callback = None
303 self.thread = None
304
305 def start(self, callback=None):
306 """Simulate starting the background thread."""
307 self.callback = callback
308 self.thread = Mock()
309 self.thread.is_alive.return_value = True
310 if callback:
311 all_listeners.append(self)
312
313 def emit(self, tract, event, **kwargs):
314 """Emit event and broadcast to all listeners."""
315 # Return False if not started yet (matches real behavior)
316 if self.thread is None or not self.thread.is_alive():
317 return False
318
319 # Build message
320 msg = {"tract": tract, "event": event, **kwargs}
321 if "ts" not in msg:
322 msg["ts"] = now_ms()
323
324 # Broadcast to all listeners
325 for listener in all_listeners:
326 if listener.callback:
327 listener.callback(msg)
328
329 return True
330
331 def stop(self):
332 """Stop connection and remove from listeners."""
333 if self in all_listeners:
334 all_listeners.remove(self)
335 self.thread = None
336 self.callback = None
337
338 # Patch both import locations
339 monkeypatch.setattr("think.runner.CallosumConnection", MockCallosumConnection)
340 monkeypatch.setattr("think.callosum.CallosumConnection", MockCallosumConnection)
341 monkeypatch.setattr("think.supervisor.CallosumConnection", MockCallosumConnection)
342
343
344def setup_google_genai_stub(monkeypatch, *, with_thinking=False):
345 """Set up a complete Google GenAI stub for testing.
346
347 Args:
348 monkeypatch: pytest monkeypatch fixture
349 with_thinking: If True, mock responses include thinking parts
350
351 Returns:
352 The DummyChat class for inspection if needed
353 """
354 from types import SimpleNamespace
355
356 google_mod = types.ModuleType("google")
357 genai_mod = types.ModuleType("google.genai")
358 errors_mod = types.ModuleType("google.genai.errors")
359
360 # Error classes matching actual SDK structure
361 class APIError(Exception):
362 pass
363
364 class ServerError(APIError):
365 pass
366
367 class ClientError(APIError):
368 pass
369
370 errors_mod.APIError = APIError
371 errors_mod.ServerError = ServerError
372 errors_mod.ClientError = ClientError
373
374 class DummyChat:
375 """Mock chat that optionally returns thinking parts."""
376
377 kwargs = None # Class var to capture last call for inspection
378
379 def __init__(self, model, history=None, config=None):
380 self.model = model
381 self.history = list(history or [])
382 self.config = config
383
384 def get_history(self):
385 return list(self.history)
386
387 def record_history(self, content):
388 self.history.append(content)
389
390 async def send_message(self, message, config=None):
391 DummyChat.kwargs = {
392 "message": message,
393 "config": config,
394 "model": self.model,
395 }
396 if with_thinking:
397 # Response with thinking parts matching actual SDK structure
398 thinking_part = SimpleNamespace(
399 thought=True,
400 text="I need to analyze this step by step.",
401 )
402 answer_part = SimpleNamespace(
403 thought=False,
404 text="ok",
405 )
406 candidate = SimpleNamespace(
407 content=SimpleNamespace(parts=[thinking_part, answer_part]),
408 )
409 return SimpleNamespace(text="ok", candidates=[candidate])
410 else:
411 # Simple response without thinking
412 return SimpleNamespace(text="ok")
413
414 class DummyChats:
415 def create(self, *, model, config=None, history=None):
416 return DummyChat(model, history=history, config=config)
417
418 class DummyModels:
419 """Mock for client.models.generate_content (non-chat generate API)."""
420
421 def generate_content(self, *, model, contents, config=None):
422 return SimpleNamespace(text="[]", candidates=[], usage_metadata=None)
423
424 class DummyClient:
425 def __init__(self, *a, **k):
426 self.chats = DummyChats()
427 self.models = DummyModels()
428 self.aio = SimpleNamespace(chats=DummyChats(), models=DummyModels())
429
430 genai_mod.Client = DummyClient
431 genai_mod.errors = errors_mod
432 genai_mod.types = SimpleNamespace(
433 GenerateContentConfig=lambda **k: SimpleNamespace(**k),
434 ToolConfig=lambda **k: SimpleNamespace(**k),
435 FunctionCallingConfig=lambda **k: SimpleNamespace(**k),
436 ThinkingConfig=lambda **k: SimpleNamespace(**k),
437 Content=lambda **k: SimpleNamespace(**k),
438 Part=lambda **k: SimpleNamespace(**k),
439 HttpOptions=lambda **k: SimpleNamespace(**k),
440 HttpRetryOptions=lambda **k: SimpleNamespace(**k),
441 )
442 google_mod.genai = genai_mod
443 monkeypatch.setitem(sys.modules, "google", google_mod)
444 monkeypatch.setitem(sys.modules, "google.genai", genai_mod)
445 monkeypatch.setitem(sys.modules, "google.genai.errors", errors_mod)
446
447 return DummyChat