90 lines
3.2 KiB
Python
90 lines
3.2 KiB
Python
from __future__ import annotations
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
sys.path.insert(0, str(ROOT / "scripts"))
|
|
|
|
import search_1c_semantic_cache as semantic_search # noqa: E402
|
|
|
|
|
|
def test_search_semantic_cache_sends_query_embedding_and_validation(monkeypatch) -> None:
|
|
seen: dict = {}
|
|
|
|
def fake_adapter_call(adapter_url: str, method: str, payload: dict, *, timeout_seconds: int = 180) -> dict:
|
|
seen["adapter_url"] = adapter_url
|
|
seen["method"] = method
|
|
seen["payload"] = payload
|
|
return {
|
|
"status": "ok",
|
|
"matches": [
|
|
{
|
|
"document_id": "doc-1",
|
|
"score": 0.9,
|
|
"match_by": "vector_embedding",
|
|
"object": {"kind": "Template", "name": "ПФ_MXL"},
|
|
"cache": {"embedding_model": "local-hashing-v1", "vector_status": "embedded"},
|
|
}
|
|
],
|
|
}
|
|
|
|
monkeypatch.setattr(semantic_search, "adapter_call", fake_adapter_call)
|
|
|
|
result = semantic_search.search_semantic_cache(
|
|
adapter_url="http://adapter/rpc",
|
|
base_id="upo_test",
|
|
query="ОбластьШапка",
|
|
kind="Template",
|
|
limit=3,
|
|
dimensions=12,
|
|
validate_candidates=True,
|
|
validation_limit=2,
|
|
)
|
|
|
|
assert result["status"] == "ok"
|
|
assert result["client_embedding"]["dimensions"] == 12
|
|
assert seen["method"] == "semantic.cache.search"
|
|
assert seen["payload"]["base_id"] == "upo_test"
|
|
assert seen["payload"]["query"] == "ОбластьШапка"
|
|
assert seen["payload"]["kind"] == "Template"
|
|
assert seen["payload"]["limit"] == 3
|
|
assert seen["payload"]["validate_candidates"] is True
|
|
assert seen["payload"]["validation_limit"] == 2
|
|
assert isinstance(seen["payload"]["query_embedding"], list)
|
|
assert len(seen["payload"]["query_embedding"]) == 12
|
|
|
|
|
|
def test_search_semantic_cache_can_embed_pending_before_search(monkeypatch) -> None:
|
|
calls: list[str] = []
|
|
|
|
def fake_embed_pending_semantic_cache(**kwargs) -> dict:
|
|
calls.append("embed")
|
|
assert kwargs["base_id"] == "upo_test"
|
|
assert kwargs["kind"] == "Template"
|
|
assert kwargs["limit"] == 7
|
|
assert kwargs["batch_size"] == 3
|
|
return {"status": "ok", "counts": {"pending": 1, "processed": 1, "stored": 1}, "embedding": {"model": "local-hashing-v1"}}
|
|
|
|
def fake_adapter_call(adapter_url: str, method: str, payload: dict, *, timeout_seconds: int = 180) -> dict:
|
|
calls.append(method)
|
|
return {"status": "ok", "matches": []}
|
|
|
|
monkeypatch.setattr(semantic_search, "embed_pending_semantic_cache", fake_embed_pending_semantic_cache)
|
|
monkeypatch.setattr(semantic_search, "adapter_call", fake_adapter_call)
|
|
|
|
result = semantic_search.search_semantic_cache(
|
|
adapter_url="http://adapter/rpc",
|
|
base_id="upo_test",
|
|
query="ОбластьШапка",
|
|
kind="Template",
|
|
dimensions=8,
|
|
embed_pending=True,
|
|
embed_limit=7,
|
|
embed_batch_size=3,
|
|
)
|
|
|
|
assert calls == ["embed", "semantic.cache.search"]
|
|
assert result["embedding_refresh"]["counts"]["stored"] == 1
|