import hashlib import math import sys from dataclasses import dataclass from pathlib import Path import pytest # The modules under test live in the sibling scripts/ directory. sys.path.insert(0, str(Path(__file__).parent.parent / "scripts")) import wiki_config import wiki_search import wiki_sync from wiki_db import EMBEDDING_DIMS CONFIG_TEMPLATE = """ embedding: endpoint: http://embed.invalid:11434 model: qwen3-embedding:0.6b dims: 1024 batch: 4 keep_alive: -1 query_prefix: "Instruct: task\\nQuery: " sources: {sources} """ WORKSPACE_SOURCE = """ workspace: kind: workspace paths: - "notes/**" - "develop/**" include: ["*.md"] exclude: - "**/inbox/**" - "develop/history.md" """ @dataclass class WikiEnv: workspace: Path wiki_dir: Path config_path: Path db_path: Path log_path: Path def write_config(self, sources: str = WORKSPACE_SOURCE) -> None: self.config_path.write_text(CONFIG_TEMPLATE.format(sources=sources), encoding="utf-8") def write_file(self, rel_path: str, text: str) -> Path: path = self.workspace / rel_path path.parent.mkdir(parents=True, exist_ok=True) path.write_text(text, encoding="utf-8") return path def log_text(self) -> str: return self.log_path.read_text(encoding="utf-8") if self.log_path.exists() else "" @pytest.fixture def wiki_env(tmp_path, monkeypatch) -> WikiEnv: """Redirect every wiki path at a throwaway workspace.""" workspace = tmp_path / "workspace" wiki_dir = workspace / "wiki" wiki_dir.mkdir(parents=True) env = WikiEnv( workspace=workspace, wiki_dir=wiki_dir, config_path=wiki_dir / "config.yaml", db_path=wiki_dir / "index.sqlite", log_path=workspace / "log" / "wiki_sync.log", ) monkeypatch.setenv("WIKI_DB", str(env.db_path)) monkeypatch.setenv("WIKI_CONFIG", str(env.config_path)) monkeypatch.setattr(wiki_config, "WORKSPACE", workspace) monkeypatch.setattr(wiki_config, "WIKI_DIR", wiki_dir) monkeypatch.setattr(wiki_config, "REMOTE_DIR", wiki_dir / "remote") monkeypatch.setattr(wiki_config, "LOCK_PATH", wiki_dir / ".sync.lock") monkeypatch.setattr(wiki_sync, "WIKI_DIR", wiki_dir) monkeypatch.setattr(wiki_sync, "LOCK_PATH", wiki_dir / ".sync.lock") monkeypatch.setattr(wiki_sync, "SYNC_LOG_PATH", env.log_path) env.write_config() return env class FakeEmbedder: """Bag-of-words vectors: cosine tracks lexical overlap, so ranks are predictable. Enough to exercise the KNN and RRF plumbing without a live model. Semantic quality is measured against the real endpoint, not here. """ def __init__(self, config): self.config = config self.calls: list[list[str]] = [] def embed_documents(self, texts): self.calls.append(list(texts)) return [self._vector(text) for text in texts] def embed_query(self, text): return self._vector(self.config.query_prefix + text) def probe(self): return None def _vector(self, text: str) -> list[float]: vector = [0.0] * EMBEDDING_DIMS for word in _words(text): digest = hashlib.sha256(word.encode("utf-8")).digest() vector[int.from_bytes(digest[:4], "big") % EMBEDDING_DIMS] += 1.0 norm = math.sqrt(sum(value * value for value in vector)) if norm == 0.0: vector[0] = 1.0 return vector return [value / norm for value in vector] def _words(text: str) -> list[str]: return [word for word in "".join(c.lower() if c.isalnum() else " " for c in text).split() if len(word) > 2] @pytest.fixture def fake_embedder(monkeypatch): """Swap the Ollama client for the deterministic fake in both entry points.""" created: list[FakeEmbedder] = [] def factory(config): embedder = FakeEmbedder(config) created.append(embedder) return embedder monkeypatch.setattr(wiki_sync, "OllamaEmbedder", factory) monkeypatch.setattr(wiki_search, "OllamaEmbedder", factory) return created