File size: 4,237 Bytes
f9ba10a | 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 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 | """端到端集成测试."""
import tempfile
import os
import pytest
from hermes_core.types import HERMES_DATA_DIR as _ORIG_DIR
from hermes_core.querier import HermesClient
pytestmark = pytest.mark.requires_embedder
@pytest.fixture(autouse=True)
def temp_data_dir(monkeypatch):
"""将 HERMES_DATA_DIR 重定向到临时目录。"""
tmp = tempfile.mkdtemp()
monkeypatch.setattr("hermes_core.types.HERMES_DATA_DIR", type(_ORIG_DIR)(tmp))
# Also patch all modules that import HERMES_DATA_DIR
import hermes_core.db as db_module
import hermes_core.trainer as trainer_module
monkeypatch.setattr(db_module, "HERMES_DATA_DIR", type(_ORIG_DIR)(tmp))
monkeypatch.setattr(trainer_module, "HERMES_DATA_DIR", type(_ORIG_DIR)(tmp))
yield tmp
import shutil
shutil.rmtree(tmp, ignore_errors=True)
class TestE2EFlow:
def test_full_flow_record_query_refine(self):
"""完整链路：record → query → refine → query。"""
client = HermesClient(user_id="u_e2e", agent_id="test-agent")
# Step 1: 记录第二条（2 维）
r1 = client.record(
"后端API开发",
[
{"key": "language", "value": "TypeScript", "context": "默认"},
{"key": "framework", "value": "Express", "context": "默认"},
],
)
assert r1["status"] == "recorded"
# Step 2: 语义相似的 query 应匹配到 scope
result = client.query("帮我写一个后端API服务")
assert result.matched_scope is not None
assert any(p.key == "language" and p.value == "TypeScript"
for p in result.related_preferences)
# Step 3: refine —— narrow
from hermes_core.refiner import refine_scene
from hermes_core.embedder import Embedder
embedder = Embedder()
ref = refine_scene("u_e2e", r1["id"], "Express REST API开发",
"narrow", embedder)
assert ref["status"] == "refined"
# Step 4: query 现在应匹配到细化的 scope
result2 = client.query("Express REST API服务端开发")
assert result2.matched_scope is not None
def test_dimension_constraint_blocks_lower_dim(self):
"""维度约束：低维记录应被拒绝。"""
client = HermesClient(user_id="u_e2e_dim", agent_id="test")
r1 = client.record(
"数据处理",
[
{"key": "language", "value": "Python", "context": ""},
{"key": "lib", "value": "Pandas", "context": ""},
{"key": "style", "value": "functional", "context": ""},
],
)
assert r1["status"] == "recorded"
r2 = client.record(
"数据处理",
[{"key": "language", "value": "R", "context": ""}],
)
assert r2["status"] == "rejected"
assert "dimension" in r2["reason"].lower()
def test_conversation_id_relaxes_constraint(self):
"""同会话内维度约束松弛。"""
client = HermesClient(user_id="u_e2e_conv", agent_id="test")
conv_id = "conv_e2e_001"
r1 = client.record(
"前端开发",
[
{"key": "framework", "value": "React", "context": ""},
{"key": "language", "value": "TypeScript", "context": ""},
],
conversation_id=conv_id,
)
assert r1["status"] == "recorded"
r2 = client.record(
"前端开发",
[{"key": "testing", "value": "Vitest", "context": ""}],
conversation_id=conv_id,
)
assert r2["status"] == "recorded" # 同对话，不拒绝
def test_multiple_scopes_created_automatically(self):
"""不同话题自动创建不同 scope。"""
client = HermesClient(user_id="u_e2e_multi", agent_id="test")
r1 = client.record("后端API开发", [{"key": "lang", "value": "TS", "context": ""}])
r2 = client.record("周末去哪玩", [{"key": "pref", "value": "户外", "context": ""}])
assert r1["scope_id"] != r2["scope_id"] # 不同 scope
assert r1["status"] == "recorded"
assert r2["status"] == "recorded"
|