File size: 2,263 Bytes
767b8b9
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
"""测试 refine_scene."""

import uuid
import pytest
from hermes_core.embedder import Embedder
from hermes_core.db import init_db, upsert_scope, insert_record
from hermes_core.types import Record, Scope, Dimension
from hermes_core.refiner import refine_scene

pytestmark = pytest.mark.requires_embedder


@pytest.fixture
def embedder():
    return Embedder()


def _uid():
    return f"u_{uuid.uuid4().hex[:8]}"


def test_refine_narrow_creates_new_scope(embedder):
    uid = _uid()
    conn = init_db(uid)
    s = Scope(id="scope_a", label="后端开发",
              centroid=embedder.encode("后端开发"), record_count=1)
    upsert_scope(conn, s)
    insert_record(conn, Record(
        id=f"rec_{uuid.uuid4().hex[:8]}", user_id=uid, scope_id="scope_a",
        scope_label="后端开发",
        dimensions=[Dimension(key="language", value="Java", context="")],
    ))
    rid = conn.execute("SELECT id FROM records LIMIT 1").fetchone()["id"]
    conn.close()

    result = refine_scene(
        user_id=uid, record_id=rid,
        scope_desc="游戏Mod开发",
        direction="narrow", embedder=embedder,
    )
    assert result["status"] == "refined"
    assert result["new_scope_id"] != result["old_scope_id"]  # 新 scope


def test_refine_broaden_reassigns_to_existing(embedder):
    uid = _uid()
    conn = init_db(uid)
    s1 = Scope(id="scope_broad", label="软件开发",
               centroid=embedder.encode("软件开发"), record_count=3)
    s2 = Scope(id="scope_narrow", label="React前端",
               centroid=embedder.encode("React前端"), record_count=1)
    upsert_scope(conn, s1)
    upsert_scope(conn, s2)
    insert_record(conn, Record(
        id=f"rec_{uuid.uuid4().hex[:8]}", user_id=uid, scope_id="scope_narrow",
        scope_label="React前端",
        dimensions=[Dimension(key="framework", value="React", context="")],
    ))
    rid = conn.execute("SELECT id FROM records LIMIT 1").fetchone()["id"]
    conn.close()

    result = refine_scene(
        user_id=uid, record_id=rid,
        scope_desc="软件开发",
        direction="broaden", embedder=embedder,
    )
    assert result["status"] == "refined"
    # broaden 应被重新分配到 scope_broad
    assert result["new_scope_id"] == "scope_broad"