MaiBot/pytests/test_maisaka_builtin_query_person_profile.py at f5663ac6b6f946b07052d9b1c708971037d1c028 · A-Dawn/MaiBot · GitHub

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
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
from types import SimpleNamespace
from typing import Any, Dict

import pytest

from src.core.tooling import ToolInvocation
from src.maisaka.builtin_tool import (
    get_all_builtin_tool_specs,
    query_person_profile as query_person_profile_tool,
)
from src.maisaka.builtin_tool.context import BuiltinToolRuntimeContext


def _build_tool_ctx() -> BuiltinToolRuntimeContext:
    runtime = SimpleNamespace(
        session_id="session-1",
        chat_stream=SimpleNamespace(platform="qq", user_id="alice", group_id=""),
        log_prefix="[session-1]",
    )
    return BuiltinToolRuntimeContext(engine=SimpleNamespace(), runtime=runtime)


def _build_invocation(arguments: Dict[str, Any]) -> ToolInvocation:
    return ToolInvocation(
        tool_name="query_person_profile",
        arguments=dict(arguments),
        call_id="call-query-person-profile",
    )


@pytest.mark.asyncio
async def test_query_person_profile_requires_identifier() -> None:
    result = await query_person_profile_tool.handle_tool(
        _build_tool_ctx(),
        _build_invocation({}),
    )

    assert result.success is False
    assert "person_id 或 person_name" in result.error_message


@pytest.mark.asyncio
async def test_query_person_profile_supports_person_id(monkeypatch: pytest.MonkeyPatch) -> None:
    captured: Dict[str, Any] = {}

    async def fake_profile_admin(*, action: str, **kwargs: Any) -> Dict[str, Any]:
        captured["action"] = action
        captured["kwargs"] = dict(kwargs)
        return {
            "success": True,
            "person_id": "pid-alice",
            "person_name": "Alice",
            "profile_text": "Alice 喜欢咖啡。",
            "profile_source": "auto_snapshot",
            "has_manual_override": False,
            "evidence": [{"content": "不应返回"}],
        }

    monkeypatch.setattr(query_person_profile_tool.memory_service, "profile_admin", fake_profile_admin)

    result = await query_person_profile_tool.handle_tool(
        _build_tool_ctx(),
        _build_invocation({"person_id": "pid-alice", "limit": 4}),
    )

    assert result.success is True
    assert captured == {
        "action": "query",
        "kwargs": {"person_id": "pid-alice", "limit": 4},
    }
    assert result.content == "Alice 喜欢咖啡。"
    assert isinstance(result.structured_content, dict)
    assert result.structured_content["person_id"] == "pid-alice"
    assert result.structured_content["person_name"] == "Alice"
    assert result.structured_content["summary"] == "Alice 喜欢咖啡。"
    assert "evidence" not in result.structured_content


@pytest.mark.asyncio
async def test_query_person_profile_supports_person_name(monkeypatch: pytest.MonkeyPatch) -> None:
    captured: Dict[str, Any] = {}

    async def fake_profile_admin(*, action: str, **kwargs: Any) -> Dict[str, Any]:
        captured["action"] = action
        captured["kwargs"] = dict(kwargs)
        return {
            "success": True,
            "person_id": "pid-bob",
            "person_name": "Bob",
            "summary": "- Bob 常聊游戏\n- Bob 晚上更活跃",
            "profile_source": "manual_override",
            "has_manual_override": True,
        }

    monkeypatch.setattr(query_person_profile_tool.memory_service, "profile_admin", fake_profile_admin)

    result = await query_person_profile_tool.handle_tool(
        _build_tool_ctx(),
        _build_invocation({"person_name": "Bob", "limit": 30}),
    )

    assert result.success is True
    assert captured == {
        "action": "query",
        "kwargs": {"person_keyword": "Bob", "limit": 20},
    }
    assert result.structured_content["person_id"] == "pid-bob"
    assert result.structured_content["profile_source"] == "manual_override"
    assert result.structured_content["has_manual_override"] is True
    assert result.structured_content["traits"] == ["Bob 常聊游戏", "Bob 晚上更活跃"]


def test_builtin_tool_list_contains_query_memory_and_profile_when_enabled(monkeypatch: pytest.MonkeyPatch) -> None:
    from src.maisaka import builtin_tool as builtin_tool_module

    monkeypatch.setattr(
        builtin_tool_module,
        "global_config",
        SimpleNamespace(
            a_memorix=SimpleNamespace(
                integration=SimpleNamespace(
                    enable_memory_query_tool=True,
                    enable_person_profile_query_tool=True,
                )
            )
        ),
    )

    specs = get_all_builtin_tool_specs()
    names = {spec.name for spec in specs if spec.enabled}

    assert "query_memory" in names
    assert "query_person_profile" in names


def test_builtin_tool_list_respects_profile_query_config(monkeypatch: pytest.MonkeyPatch) -> None:
    from src.maisaka import builtin_tool as builtin_tool_module

    monkeypatch.setattr(
        builtin_tool_module,
        "global_config",
        SimpleNamespace(
            a_memorix=SimpleNamespace(
                integration=SimpleNamespace(
                    enable_memory_query_tool=True,
                    enable_person_profile_query_tool=False,
                )
            )
        ),
    )

    specs = get_all_builtin_tool_specs()
    profile_spec = next(spec for spec in specs if spec.name == "query_person_profile")

    assert profile_spec.enabled is False