142 lines
4.0 KiB
Python
142 lines
4.0 KiB
Python
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
|