RRRosmontis
34089d0042
Include utility modules, shared channels, prompts, read-only skills, wardrobe memory and regression tests. Externalize legacy credentials and exclude private runtime state.
124 lines
7.1 KiB
Python
124 lines
7.1 KiB
Python
"""Private wardrobe tool, scope and prompt integration; no network or real state."""
|
|
import json
|
|
from pathlib import Path
|
|
import sys
|
|
import tempfile
|
|
from types import SimpleNamespace
|
|
import unittest
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
BASE = Path(__file__).resolve().parents[1]
|
|
sys.path[:0] = [str(BASE / 'bot-utility'), str(BASE)]
|
|
import wardrobe as module
|
|
from agent import ToolContext, ToolExecutor
|
|
from chat import engine
|
|
|
|
|
|
class WardrobeIntegrationTests(unittest.IsolatedAsyncioTestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.store = module.WardrobeStore(Path(self.tmp.name) / 'wardrobe.db')
|
|
self.channel = SimpleNamespace(name='tg', is_admin=lambda uid: False)
|
|
self.ctx = ToolContext(self.channel, 'private', 1001, 123)
|
|
self.other = ToolContext(self.channel, 'private', 1002, 456)
|
|
wrapped = SimpleNamespace(wardrobe=self.store, format_item=module.format_item)
|
|
self.executor = ToolExecutor(artifacts=MagicMock(), modules={'wardrobe': wrapped})
|
|
|
|
def tearDown(self):
|
|
if hasattr(self.store, 'close'):
|
|
self.store.close()
|
|
self.tmp.cleanup()
|
|
|
|
async def test_add_and_read_full_description_and_other_owner_denied(self):
|
|
description = '一条蓝色的、底部带一条白色环绕条纹的短裙。'
|
|
result = await self.executor.execute('wardrobe', {'action': 'add', 'params': {
|
|
'name': '蓝色短裙', 'category': '短裙', 'description': description}}, self.ctx)
|
|
self.assertTrue(result.ok, result.safe_message)
|
|
item = result.data['item']
|
|
self.assertEqual(item['description'], description)
|
|
read = await self.executor.execute('wardrobe', {'action': 'list', 'params': {'query': '蓝色'}}, self.ctx)
|
|
self.assertEqual(read.data['items'][0]['description'], description)
|
|
failed = await self.executor.execute('wardrobe', {'action': 'delete', 'params': {'id': item['id']}}, self.other)
|
|
self.assertFalse(failed.ok)
|
|
self.assertIn('蓝色短裙', self.store.context_for_chat(1001))
|
|
self.assertNotIn('蓝色短裙', self.store.context_for_chat(1002))
|
|
|
|
async def test_tool_receipts_never_expose_record_json_or_metadata(self):
|
|
added = await self.executor.execute('wardrobe', {'action': 'add', 'params': {
|
|
'name': '蓝色短裙', 'description': '白色环绕条纹'}}, self.ctx)
|
|
item_id = added.data['item']['id']
|
|
updated = await self.executor.execute('wardrobe', {'action': 'update', 'params': {
|
|
'id': item_id, 'description': '裙摆底部白色环绕条纹'}}, self.ctx)
|
|
listed = await self.executor.execute('wardrobe', {'action': 'list', 'params': {}}, self.ctx)
|
|
deleted = await self.executor.execute('wardrobe', {'action': 'delete', 'params': {'id': item_id}}, self.ctx)
|
|
for result in (added, updated, listed, deleted):
|
|
self.assertTrue(result.ok, result.safe_message)
|
|
self.assertNotIn('{', result.safe_message)
|
|
self.assertNotIn(item_id, result.safe_message)
|
|
self.assertNotIn('uid', result.safe_message)
|
|
for item in result.data.get('items', [result.data.get('item', {})]):
|
|
self.assertNotIn('uid', item)
|
|
self.assertNotIn('created_at', item)
|
|
self.assertNotIn('updated_at', item)
|
|
# Storage still retains full records; only outward receipts were changed.
|
|
persisted = self.store.list(self.ctx)[0]
|
|
self.assertIn('uid', persisted)
|
|
|
|
async def test_group_cannot_list_or_mutate_even_as_admin(self):
|
|
group = ToolContext(self.channel, 'group', 1001, 999, is_admin=True)
|
|
self.assertNotIn('wardrobe', [x['name'] for x in self.executor.prompt_schema(group)])
|
|
self.assertIn('wardrobe', [x['name'] for x in self.executor.prompt_schema(self.ctx)])
|
|
for action, params in [('list', {}), ('add', {'name': '袜子'}),
|
|
('update', {'id': 'default-white-dress', 'name': 'x'}),
|
|
('delete', {'id': 'default-white-dress'})]:
|
|
result = await self.executor.execute('wardrobe', {'action': action, 'params': params}, group)
|
|
self.assertEqual(result.code, 'permission_denied')
|
|
|
|
async def test_model_cannot_override_owner(self):
|
|
result = await self.executor.execute('wardrobe', {'action': 'add', 'params': {
|
|
'name': '袜子', 'uid': 1002}}, self.ctx)
|
|
self.assertEqual(result.code, 'invalid_args')
|
|
|
|
async def test_list_pages_preserve_whole_records(self):
|
|
for n in range(12):
|
|
self.store.add(self.ctx, f'衣物{n}', '细节' * 900, '其他')
|
|
first = await self.executor.execute('wardrobe', {'action': 'list', 'params': {'limit': 20}}, self.ctx)
|
|
self.assertTrue(first.ok)
|
|
self.assertGreater(first.data['next_offset'], 0)
|
|
self.assertLess(first.data['next_offset'], first.data['total'])
|
|
for item in first.data['items']:
|
|
if item['id'] != 'default-white-dress': self.assertEqual(item['description'], '细节' * 900)
|
|
second = await self.executor.execute('wardrobe', {'action': 'list', 'params': {
|
|
'offset': first.data['next_offset']}}, self.ctx)
|
|
ids = {x['id'] for x in first.data['items']}
|
|
self.assertFalse(ids.intersection(x['id'] for x in second.data['items']))
|
|
|
|
def test_private_and_proactive_prompt_include_wardrobe_after_stable_instructions(self):
|
|
with patch.object(engine, 'user_persona_prompt', return_value='PERSONA'), \
|
|
patch.object(engine, 'load_user_status', return_value={'pause': False, 'sleep': False}), \
|
|
patch.object(engine.memory_store, 'get_relation', return_value=None), \
|
|
patch.object(engine.memory_store, 'list_entities', return_value=[]), \
|
|
patch.object(engine.memory_store, 'recall_facts', return_value=[]), \
|
|
patch.object(engine.memory_store, 'migrate_legacy'), \
|
|
patch.object(engine, 'get_history_within_budget', return_value=[]), \
|
|
patch.object(engine, 'get_last_active', return_value=0), \
|
|
patch.object(engine, '_sticker_prompt_summary', return_value=''), \
|
|
patch.object(engine.notebook.notebook, 'context_for_chat', return_value=''), \
|
|
patch.object(engine.wardrobe, 'wardrobe', self.store):
|
|
for is_auto in (False, True):
|
|
prompt = engine.build_system_prompt(1001, is_auto=is_auto)
|
|
self.assertIn('白色连衣裙', prompt)
|
|
self.assertLess(prompt.index('当你回复时'), prompt.index('白色连衣裙'))
|
|
self.assertNotIn('现在是 ', prompt)
|
|
|
|
async def test_group_prompt_never_reads_private_wardrobe(self):
|
|
with patch.object(engine, 'get_recent_group_history', return_value=[]), \
|
|
patch.object(engine, 'call_deepseek', new=AsyncMock(return_value=json.dumps({'message': [], 'tools': []}))), \
|
|
patch.object(engine.wardrobe.wardrobe, 'context_for_chat') as context:
|
|
await engine._call_group_model_and_reply_impl(None, 999, '你好', 0)
|
|
context.assert_not_called()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|