53 lines
1.9 KiB
Python
53 lines
1.9 KiB
Python
"""Tests for app/context/embedder.py — graceful no-op without sqlite-vec."""
|
|
import sqlite3
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
import pytest
|
|
from app.context import embedder as emb_mod
|
|
|
|
|
|
@pytest.fixture
|
|
def db(tmp_path):
|
|
db_path = tmp_path / "t.db"
|
|
conn = sqlite3.connect(str(db_path))
|
|
conn.executescript("""
|
|
CREATE TABLE context_documents (
|
|
id TEXT PRIMARY KEY, filename TEXT NOT NULL, doc_type TEXT NOT NULL,
|
|
full_text TEXT NOT NULL, file_size INTEGER, uploaded_at TEXT NOT NULL
|
|
);
|
|
CREATE TABLE context_chunks (
|
|
id TEXT PRIMARY KEY, document_id TEXT NOT NULL
|
|
REFERENCES context_documents(id) ON DELETE CASCADE,
|
|
chunk_index INTEGER NOT NULL, text TEXT NOT NULL, embedding BLOB
|
|
);
|
|
INSERT INTO context_documents VALUES ('d1','test.md','markdown','hello',5,'2026-01-01T00:00:00+00:00');
|
|
INSERT INTO context_chunks VALUES ('c1','d1',0,'hello world',NULL);
|
|
""")
|
|
conn.commit()
|
|
conn.close()
|
|
return db_path
|
|
|
|
|
|
def test_embed_skipped_when_extension_absent(db):
|
|
with patch.object(emb_mod, "EMBEDDING_AVAILABLE", False):
|
|
count = emb_mod.embed_chunks(db, "d1", "http://localhost:11434")
|
|
assert count == 0
|
|
|
|
|
|
def test_embed_calls_ollama_when_available(db):
|
|
import httpx
|
|
|
|
class FakeResponse:
|
|
status_code = 200
|
|
def raise_for_status(self): pass
|
|
def json(self): return {"embedding": [0.1, 0.2, 0.3]}
|
|
|
|
with patch.object(emb_mod, "EMBEDDING_AVAILABLE", True), \
|
|
patch("app.context.embedder.httpx.post", return_value=FakeResponse()):
|
|
count = emb_mod.embed_chunks(db, "d1", "http://localhost:11434")
|
|
assert count == 1
|
|
# Verify blob was written
|
|
conn = sqlite3.connect(str(db))
|
|
row = conn.execute("SELECT embedding FROM context_chunks WHERE id='c1'").fetchone()
|
|
conn.close()
|
|
assert row[0] is not None
|