note-zero/tests/test_db.py

from pathlib import Path

from app.config import UserSeed
from app.db import Database


ADMIN = UserSeed(
    id="admin-user",
    label="管理员",
    role="admin",
    access_key_hash="hash-admin",
)
MEMBER = UserSeed(
    id="member-user",
    label="用户二",
    role="member",
    access_key_hash="hash-member",
)


def database(tmp_path: Path) -> Database:
    db = Database(tmp_path / "test.sqlite3")
    db.initialize([ADMIN, MEMBER])
    return db


def assessment(idea_id: str | None = None, maturity: float = 38):
    return {
        "idea_id": idea_id,
        "title": "无分类的笔记入口",
        "summary": "先保留念头，再让结构在后台形成。",
        "maturity": maturity,
        "confidence": 0.76,
        "motion": "从抱怨聚成原则",
        "position": "已经有清楚的交互原则，还没有碰到真实使用。",
        "tension": "自由输入与系统内部结构之间需要同时成立。",
        "trajectory": "从输入摩擦的感受，收束成了产品边界。",
        "possible_moves": ["做一个只保留原文的输入原型"],
        "relevance": 1,
    }


def test_fragment_is_saved_before_analysis(tmp_path: Path):
    db = database(tmp_path)

    fragment = db.create_fragment(ADMIN.id, "一个还没有分类的念头")

    assert db.list_fragments(ADMIN.id) == [
        {
            "id": fragment["id"],
            "content": "一个还没有分类的念头",
            "created_at": fragment["created_at"],
        }
    ]
    assert db.get_fragment(fragment["id"], ADMIN.id)["analysis_status"] == "pending"


def test_user_data_is_isolated_at_query_layer(tmp_path: Path):
    db = database(tmp_path)
    admin_fragment = db.create_fragment(ADMIN.id, "管理员的私密念头")
    member_fragment = db.create_fragment(MEMBER.id, "用户二的私密念头")
    db.apply_assessments(ADMIN.id, admin_fragment["id"], [assessment()])
    admin_idea = db.list_ideas(ADMIN.id)[0]

    assert [item["content"] for item in db.list_fragments(ADMIN.id)] == [
        "管理员的私密念头"
    ]
    assert [item["content"] for item in db.list_fragments(MEMBER.id)] == [
        "用户二的私密念头"
    ]
    assert db.get_fragment(member_fragment["id"], ADMIN.id) is None
    assert db.get_idea(MEMBER.id, admin_idea["id"]) is None
    assert db.update_idea_override(MEMBER.id, admin_idea["id"], 80) is None
    assert db.curator_context(MEMBER.id)[0] == []


def test_apply_assessment_builds_trajectory(tmp_path: Path):
    db = database(tmp_path)
    first = db.create_fragment(ADMIN.id, "笔记不应该要求先分类")
    db.apply_assessments(ADMIN.id, first["id"], [assessment()])
    ideas = db.list_ideas(ADMIN.id)

    assert len(ideas) == 1
    assert ideas[0]["maturity"] == 38
    assert ideas[0]["fragment_count"] == 1

    second = db.create_fragment(ADMIN.id, "AI 的判断不能出现在记录流里")
    db.apply_assessments(
        ADMIN.id,
        second["id"],
        [assessment(ideas[0]["id"], maturity=46)],
    )
    idea = db.get_idea(ADMIN.id, ideas[0]["id"])

    assert idea["fragment_count"] == 2
    assert len(idea["snapshots"]) == 2
    assert idea["maturity"] == 46


def test_manual_position_is_authoritative(tmp_path: Path):
    db = database(tmp_path)
    fragment = db.create_fragment(ADMIN.id, "一个想法")
    db.apply_assessments(ADMIN.id, fragment["id"], [assessment(maturity=30)])
    idea = db.list_ideas(ADMIN.id)[0]

    overridden = db.update_idea_override(ADMIN.id, idea["id"], 61)
    assert overridden["maturity"] == 61
    assert overridden["maturity_ai"] == 30
    assert overridden["is_overridden"] is True

    restored = db.update_idea_override(ADMIN.id, idea["id"], None)
    assert restored["maturity"] == 30
    assert restored["is_overridden"] is False


def test_existing_unowned_data_migrates_to_first_admin(tmp_path: Path):
    db = Database(tmp_path / "legacy.sqlite3")
    db.initialize()
    with db.connect() as connection:
        connection.execute(
            """
            INSERT INTO fragments (
                id, user_id, content, created_at, analysis_status
            ) VALUES ('legacy-fragment', NULL, '旧记录', '2026-01-01', 'done')
            """
        )
        connection.execute(
            """
            INSERT INTO ideas (
                id, user_id, title, summary, maturity_ai, confidence,
                motion, position, tension, trajectory, possible_moves,
                created_at, updated_at
            ) VALUES (
                'legacy-idea', NULL, '旧想法', '摘要', 10, .5,
                '浮现', '位置', '张力', '轨迹', '[]',
                '2026-01-01', '2026-01-01'
            )
            """
        )

    db.initialize([ADMIN, MEMBER])

    assert db.get_fragment("legacy-fragment", ADMIN.id)["content"] == "旧记录"
    assert db.get_idea(ADMIN.id, "legacy-idea")["title"] == "旧想法"
    assert db.get_fragment("legacy-fragment", MEMBER.id) is None


def test_agent_run_records_metrics_and_events(tmp_path: Path):
    db = database(tmp_path)
    fragment = db.create_fragment(ADMIN.id, "可观测的一次分析")
    run_id = db.start_agent_run(
        ADMIN.id,
        fragment["id"],
        "deepseek-v4-pro",
        "prompt.v1",
        "high",
    )
    db.add_agent_event(
        run_id,
        ADMIN.id,
        "tool_search_ideas",
        {"query": "可观测", "result_count": 0},
        duration_ms=3,
    )
    db.finish_agent_run(
        run_id,
        status="success",
        duration_ms=1200,
        attempt_count=1,
        model_rounds=2,
        tool_calls=1,
        search_calls=1,
        inspection_calls=0,
        input_tokens=100,
        output_tokens=50,
        reasoning_tokens=30,
        cached_tokens=20,
    )

    run = db.get_agent_run(run_id)
    events = db.list_agent_events(run_id)
    overview = db.admin_overview()

    assert run["status"] == "success"
    assert run["reasoning_tokens"] == 30
    assert [event["event_type"] for event in events] == [
        "run_started",
        "tool_search_ideas",
    ]
    assert overview["last_24h"]["success_rate"] == 100