"""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()