File size: 2,357 Bytes
62953fa | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 | """测试 query 模块."""
import tempfile, sqlite3, os
import pytest
from hermes_core.embedder import Embedder
from hermes_core.db import init_db, insert_record, upsert_scope
from hermes_core.types import Record, Scope, Dimension
from hermes_core.querier import query, HermesClient
pytestmark = pytest.mark.requires_embedder
@pytest.fixture
def embedder():
return Embedder()
def test_query_no_scopes_returns_empty(embedder):
result = query("u_test", "帮我写个API", embedder)
assert result.matched_scope is None
assert len(result.active_loras) == 0
assert len(result.related_preferences) == 0
def test_query_matches_existing_scope(embedder):
# 创建一个 scope 和相关记录
conn = init_db("u_test")
s = Scope(id="scope_a", label="后端API开发",
centroid=embedder.encode("后端API开发"), record_count=3)
upsert_scope(conn, s)
insert_record(conn, Record(
id="r1", user_id="u_test", scope_id="scope_a",
scope_label="后端API开发",
dimensions=[Dimension(key="language", value="TypeScript", context="默认")],
confidence=0.9, occurrences=3,
))
conn.close()
result = query("u_test", "帮我写一个后端API接口", embedder)
assert result.matched_scope is not None
assert result.matched_scope.scope_id == "scope_a"
assert any(p.key == "language" for p in result.related_preferences)
def test_query_no_match_when_too_different(embedder):
conn = init_db("u_test")
s = Scope(id="scope_a", label="后端API开发",
centroid=embedder.encode("后端API开发"), record_count=3)
upsert_scope(conn, s)
conn.close()
result = query("u_test", "推荐个周末去处", embedder)
# 周末去处与后端开发相似度低 → 不匹配
assert result.matched_scope is None
class TestHermesClient:
def test_client_query(self, embedder):
client = HermesClient(user_id="u_client_test", agent_id="agent-1")
result = client.query("帮我写个API")
assert isinstance(result.matched_scope, (type(None), object))
def test_client_record(self, embedder):
client = HermesClient(user_id="u_client_test", agent_id="agent-1")
r = client.record("后端开发", [{"key": "language", "value": "TS", "context": ""}])
assert r["status"] == "recorded"
|