tests/test_refiner.py

"""测试 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"