Spaces:
Running
Running
| import asyncio | |
| import copy | |
| import os | |
| from pathlib import Path | |
| import tempfile | |
| import unittest | |
| from unittest.mock import patch | |
| os.environ['GRADIO_ANALYTICS_ENABLED'] = 'False' | |
| import gradio as gr | |
| from gradio.state_holder import SessionState | |
| import app | |
| from engine import memory | |
| from engine.saveload import export_save | |
| from engine.saveload import import_save | |
| from test_conversations import Client, CHAR, OTHER | |
| class AppTests(unittest.TestCase): | |
| def setUp(self): | |
| self.temp = tempfile.TemporaryDirectory() | |
| self.addCleanup(self.temp.cleanup) | |
| self.patch = patch.object(memory, 'MEMORY_DIR', Path(self.temp.name) / 'sessions') | |
| self.patch.start() | |
| self.addCleanup(self.patch.stop) | |
| self.request = gr.Request(session_hash='browser-one') | |
| self.session = app.session_key(self.request) | |
| def test_translator_defaults_agree(self): | |
| self.assertFalse(app.use_translator_state.value) | |
| self.assertFalse(app.translator_toggle.value) | |
| def test_session_states_are_registered_in_browser_config(self): | |
| component_ids = {c['id'] for c in app.demo.get_config_file()['components']} | |
| for state in (app.char_id_state, app.persona_state, app.language_state, | |
| app.model_state, app.use_translator_state): | |
| with self.subTest(state=state._id): | |
| self.assertIn(state._id, component_ids) | |
| def test_stale_save_download_is_cleared_when_context_changes(self): | |
| fn = next(f for f in app.demo.fns.values() if f.name == 'invalidate_save_fn') | |
| self.assertEqual([app.save_file, app.save_status], fn.outputs) | |
| self.assertTrue({(app.char_selector._id, 'input'), | |
| (app.load_upload._id, 'change'), | |
| (app.chatbot.chatbot_state._id, 'change'), | |
| (app.persona_state._id, 'change')}.issubset(set(fn.targets))) | |
| update, status = app.invalidate_save_fn() | |
| self.assertIsNone(update['value']) | |
| self.assertFalse(update['visible']) | |
| self.assertEqual('', status) | |
| def test_haley_save_load_and_continue_uses_same_character(self): | |
| async def dispatch(name, inputs, state): | |
| fn_id = next(i for i, f in app.demo.fns.items() if f.name == name) | |
| return await app.demo.process_api(fn_id, inputs, state=state, | |
| request=self.request, session_hash=self.request.session_hash) | |
| async def run(): | |
| haley = 'haley_dunphy' | |
| state = SessionState(app.demo) | |
| await dispatch('on_character_change', [haley], state) | |
| history = memory.load_conversation(self.session, haley)['recent_messages'] | |
| history.extend([{'role': 'user', 'content': 'Remember our trip?'}, | |
| {'role': 'assistant', 'content': 'Of course!'}]) | |
| app.persist_visible_history(history, haley, self.request) | |
| saved = await dispatch('save_conversation_fn', [None, None], state) | |
| path = Path(saved['data'][0]['value']['path']) | |
| self.assertEqual(haley, import_save(path)['character_id']) | |
| self.assertIn('Haley Dunphy', saved['data'][1]) | |
| await dispatch('on_character_change', [CHAR], state) | |
| with tempfile.TemporaryDirectory(dir=gr.utils.get_upload_folder()) as directory: | |
| uploaded = Path(directory) / 'haley.json' | |
| uploaded.write_bytes(path.read_bytes()) | |
| loaded = await dispatch('load_conversation_fn', | |
| [{'path': str(uploaded), 'meta': {'_type': 'gradio.FileData'}}, None], state) | |
| self.assertEqual(haley, state[app.char_id_state._id]) | |
| self.assertEqual(haley, loaded['data'][11]['value']) | |
| self.assertEqual('## Haley Dunphy', loaded['data'][8]) | |
| self.assertEqual(history, memory.normalize_history(loaded['data'][7]['value'])) | |
| await dispatch('on_character_change', [CHAR], state) | |
| await dispatch('on_character_change', [haley], state) | |
| self.assertEqual(history, memory.normalize_history(state[app.chatbot.chatbot_state._id])) | |
| client = Client() | |
| with patch.object(app, 'HF_TOKEN', 'test'), patch.object(app, 'InferenceClient', return_value=client): | |
| result = await dispatch('respond', ['What happened next?', None, 100, .7, .95, | |
| None, None, None, None, None], state) | |
| fn_id = next(i for i, f in app.demo.fns.items() if f.name == 'respond') | |
| while result['is_generating']: | |
| result = await app.demo.process_api(fn_id, [], state=state, | |
| request=self.request, session_hash=self.request.session_hash, | |
| iterator=result['iterator']) | |
| self.assertIn('Haley', client.calls[0][0][0]['content']) | |
| self.assertEqual(history, memory.load_conversation(self.session, haley)['recent_messages'][:-2]) | |
| self.assertEqual(1, len(memory.load_conversation(self.session, CHAR)['recent_messages'])) | |
| Path(gr.utils.get_upload_folder()).mkdir(parents=True, exist_ok=True) | |
| asyncio.run(run()) | |
| def test_character_switch_restores_visible_and_internal_history(self): | |
| state = memory.load_conversation(self.session, CHAR) | |
| state['recent_messages'].extend([{'role': 'user', 'content': 'Remember me'}, | |
| {'role': 'assistant', 'content': 'I do.'}]) | |
| memory.save_conversation(self.session, CHAR, state) | |
| outputs = app.on_character_change(CHAR, self.request) | |
| self.assertEqual(state['recent_messages'], outputs[4]['value']) | |
| self.assertEqual(state['recent_messages'], memory.normalize_history(outputs[5])) | |
| self.assertEqual(outputs[5], outputs[6]) | |
| prefix, message = app.chatbot._pop_last_user_message(outputs[5]) | |
| self.assertEqual('Remember me', message) | |
| self.assertEqual(1, len(prefix)) | |
| second = app.on_character_change(CHAR, gr.Request(session_hash='browser-two')) | |
| self.assertEqual(1, len(second[5])) | |
| def test_load_updates_every_history_and_only_current_session(self): | |
| source_session = app.session_key(gr.Request(session_hash='source')) | |
| state = memory.load_conversation(source_session, OTHER) | |
| state['recent_messages'].extend([{'role': 'user', 'content': 'Loaded question'}, | |
| {'role': 'assistant', 'content': 'Loaded response'}]) | |
| memory.save_conversation(source_session, OTHER, state) | |
| path = export_save(OTHER, {'name': 'Alex'}, source_session) | |
| self.addCleanup(lambda: Path(path).unlink(missing_ok=True)) | |
| result = app.load_conversation_fn(path, CHAR, self.request) | |
| self.assertEqual(15, len(result)) | |
| self.assertEqual(OTHER, result[0]) | |
| self.assertEqual('Alex', result[2]) | |
| self.assertEqual(state['recent_messages'], result[7]['value']) | |
| self.assertEqual(state['recent_messages'], memory.normalize_history(result[12])) | |
| self.assertEqual(result[12], result[13]) | |
| self.assertEqual(state['recent_messages'], memory.load_conversation(self.session, OTHER)['recent_messages']) | |
| self.assertEqual(1, len(memory.load_conversation(self.session, CHAR)['recent_messages'])) | |
| def test_clear_and_stop_persist_immediately(self): | |
| visible = [{'role': 'user', 'content': 'hello'}, {'role': 'assistant', 'content': 'partial'}] | |
| app.stop_chat_fn(visible, CHAR, self.request) | |
| self.assertEqual(visible, memory.load_conversation(self.session, CHAR)['recent_messages']) | |
| self.assertEqual(([], [], []), app.clear_chat_fn(CHAR, self.request)) | |
| self.assertEqual([], memory.load_conversation(self.session, CHAR)['recent_messages']) | |
| def test_inference_errors_do_not_write_successful_exchange(self): | |
| history = memory.load_conversation(self.session, CHAR)['recent_messages'] | |
| with patch.object(app, 'HF_TOKEN', 'test'), patch.object(app, 'InferenceClient', return_value=Client(fail=True)): | |
| values = list(app.respond('Hi', history, 100, .7, .95, {}, CHAR, 'English', 'test', False, self.request)) | |
| self.assertIn('try again', values[-1]) | |
| self.assertEqual(history, memory.load_conversation(self.session, CHAR)['recent_messages']) | |
| def test_gradio_dispatch_switch_and_stream_with_session_injection(self): | |
| async def run(): | |
| state = SessionState(app.demo) | |
| switch_id = next(i for i, f in app.demo.fns.items() if f.name == 'on_character_change') | |
| await app.demo.process_api(switch_id, [OTHER], state=state, request=self.request, | |
| session_hash=self.request.session_hash) | |
| self.assertEqual(OTHER, state[app.char_id_state._id]) | |
| history = state[app.chatbot.chatbot_state._id] | |
| self.assertEqual(1, len(history)) | |
| client = Client() | |
| chat_id = next(i for i, f in app.demo.fns.items() if f.name == 'respond') | |
| with patch.object(app, 'HF_TOKEN', 'test'), patch.object(app, 'InferenceClient', return_value=client): | |
| inputs = ['Hi', None, 100, .7, .95, None, None, None, None, None] | |
| result = await app.demo.process_api(chat_id, inputs, state=state, request=self.request, | |
| session_hash=self.request.session_hash) | |
| while result['is_generating']: | |
| result = await app.demo.process_api(chat_id, inputs, state=state, request=self.request, | |
| session_hash=self.request.session_hash, | |
| iterator=result['iterator']) | |
| stored = memory.load_conversation(self.session, OTHER) | |
| self.assertEqual(3, len(stored['recent_messages'])) | |
| self.assertEqual('Hi', stored['recent_messages'][-2]['content']) | |
| self.assertIn('Jessica', client.calls[0][0][0]['content']) | |
| asyncio.run(run()) | |
| def test_suggestion_event_passes_active_language_and_persona(self): | |
| async def run(): | |
| state = SessionState(app.demo) | |
| state[app.language_state._id] = 'Turkish' | |
| state[app.persona_state._id] = {'name': 'Deniz', 'personality': 'shy'} | |
| fn_id = next(i for i, f in app.demo.fns.items() if f.name == 'get_suggestions_fn') | |
| suggestions = [{'tone': 'Meraklı', 'emoji': '?', 'text': 'Bana anlatır mısın?'}] * 4 | |
| with patch.object(app, 'HF_TOKEN', 'test'), patch.object(app, 'InferenceClient'), patch.object(app, 'generate_suggestions', return_value=suggestions) as generate: | |
| result = await app.demo.process_api(fn_id, [None] * 4, state=state, | |
| request=self.request, session_hash=self.request.session_hash) | |
| self.assertEqual('Turkish', generate.call_args.kwargs['language']) | |
| self.assertEqual('Deniz', generate.call_args.kwargs['persona']['name']) | |
| self.assertEqual(9, len(result['data'])) | |
| self.assertEqual('Bana anlatır mısın?', state[app.suggestion_texts[0]._id]) | |
| reset_id = next(i for i, f in app.demo.fns.items() if f.name == 'clear_suggestions_fn') | |
| await app.demo.process_api(reset_id, [], state=state, request=self.request, | |
| session_hash=self.request.session_hash) | |
| self.assertEqual('', state[app.suggestion_texts[0]._id]) | |
| asyncio.run(run()) | |
| def test_memory_panel_controls_dispatch_and_persist(self): | |
| async def run(): | |
| state = SessionState(app.demo) | |
| fn_id = next(i for i, f in app.demo.fns.items() if f.name == 'memory_action_fn' and f.inputs[0].value == 'remember') | |
| result = await app.demo.process_api(fn_id, [None, 'A personal fact.', None, None, None], | |
| state=state, request=self.request, session_hash=self.request.session_hash) | |
| stored = memory.load_conversation(self.session, CHAR) | |
| self.assertEqual('manual', stored['long_term_memory'][0]['origin']) | |
| self.assertEqual('A personal fact.', stored['long_term_memory'][0]['content']) | |
| self.assertEqual('Memory saved.', result['data'][-1]) | |
| identifier = stored['long_term_memory'][0]['id'] | |
| self.assertEqual('A personal fact.', app.select_memory_fn(identifier, CHAR, self.request)) | |
| edit_id = next(i for i, f in app.demo.fns.items() if f.name == 'memory_action_fn' and f.inputs[0].value == 'edit') | |
| await app.demo.process_api(edit_id, [None, 'Corrected fact.', identifier, None, None], | |
| state=state, request=self.request, session_hash=self.request.session_hash) | |
| self.assertEqual('Corrected fact.', memory.load_conversation(self.session, CHAR)['long_term_memory'][0]['content']) | |
| forget_id = next(i for i, f in app.demo.fns.items() if f.name == 'memory_action_fn' and f.inputs[0].value == 'forget') | |
| await app.demo.process_api(forget_id, [None, '', identifier, None, None], | |
| state=state, request=self.request, session_hash=self.request.session_hash) | |
| self.assertEqual([], memory.load_conversation(self.session, CHAR)['long_term_memory']) | |
| asyncio.run(run()) | |
| def test_relationship_panel_dispatch_and_restore(self): | |
| async def run(): | |
| state = SessionState(app.demo) | |
| state[app.language_state._id] = 'Turkish' | |
| fn_id = next(i for i, f in app.demo.fns.items() if f.name == 'relationship_options_fn') | |
| result = await app.demo.process_api(fn_id, [True, True, None, None], state=state, | |
| request=self.request, session_hash=self.request.session_hash) | |
| self.assertIn('Romantik yakınlık', result['data'][0]) | |
| self.assertIn('Cinsel çekim', result['data'][0]) | |
| self.assertFalse(any(c.isdigit() for c in result['data'][0])) | |
| panel = app.restore_panels_fn(CHAR, 'Turkish', self.request) | |
| self.assertEqual((True, True), panel[2:4]) | |
| other = app.restore_panels_fn(OTHER, 'Turkish', self.request) | |
| self.assertEqual((False, False), other[2:4]) | |
| asyncio.run(run()) | |
| def test_edit_last_message_updates_model_and_ui_histories(self): | |
| async def run(): | |
| state = SessionState(app.demo) | |
| stored = memory.load_conversation(self.session, CHAR) | |
| stored['recent_messages'].extend([{'role': 'user', 'content': 'Original question'}, | |
| {'role': 'assistant', 'content': 'Old answer'}]) | |
| memory.save_conversation(self.session, CHAR, stored) | |
| fn_id = next(i for i, f in app.demo.fns.items() if f.name == 'edit_last_message_fn') | |
| result = await app.demo.process_api(fn_id, [None], state=state, request=self.request, | |
| session_hash=self.request.session_hash) | |
| self.assertEqual('Original question', result['data'][3]['value']) | |
| self.assertEqual(stored['recent_messages'][:1], memory.normalize_history(state[app.chatbot.chatbot_state._id])) | |
| self.assertEqual(stored['recent_messages'][:1], memory.load_conversation(self.session, CHAR)['recent_messages']) | |
| asyncio.run(run()) | |
| def test_panel_poll_reflects_background_changes_without_model_calls(self): | |
| from engine.relationship import apply_relationship_changes | |
| stored = memory.load_conversation(self.session, CHAR) | |
| stored['recent_messages'].extend([{'role': 'user', 'content': 'Hello'}, {'role': 'assistant', 'content': 'Hello'}]) | |
| stored['relationship'] = apply_relationship_changes(stored['relationship'], | |
| [{'axis': 'trust', 'delta': 5, 'reason': 'A kept promise'}], 1, 3) | |
| memory.save_conversation(self.session, CHAR, stored) | |
| with patch.object(app, 'InferenceClient') as client: | |
| panel = app.refresh_panels_fn(CHAR, 'Turkish', None, self.request) | |
| client.assert_not_called() | |
| self.assertIn('Sana güvenmeye başlıyor.', panel[0]) | |
| def test_load_file_component_dispatch_restores_history_and_relationship(self): | |
| from engine.relationship import apply_relationship_changes | |
| async def run(): | |
| source = app.session_key(gr.Request(session_hash='upload-source')) | |
| stored = memory.load_conversation(source, OTHER) | |
| stored['recent_messages'].extend([ | |
| {'role': 'user', 'content': 'Uploaded question'}, | |
| {'role': 'assistant', 'content': 'Uploaded answer'}, | |
| ]) | |
| stored['relationship'] = apply_relationship_changes(stored['relationship'], | |
| [{'axis': 'trust', 'delta': 6, 'reason': 'A kept promise'}], 1, 3) | |
| memory.save_conversation(source, OTHER, stored) | |
| exported = export_save(OTHER, {'name': 'Uploaded persona'}, source) | |
| self.addCleanup(lambda: Path(exported).unlink(missing_ok=True)) | |
| # Match the File component's actual upload payload, including Gradio's | |
| # permitted upload directory, rather than calling the handler directly. | |
| with tempfile.TemporaryDirectory(dir=gr.utils.get_upload_folder()) as directory: | |
| uploaded = Path(directory) / 'save.json' | |
| uploaded.write_bytes(Path(exported).read_bytes()) | |
| state = SessionState(app.demo) | |
| fn_id = next(i for i, f in app.demo.fns.items() if f.name == 'load_conversation_fn') | |
| result = await app.demo.process_api(fn_id, | |
| [{'path': str(uploaded), 'meta': {'_type': 'gradio.FileData'}}, None], | |
| state=state, request=self.request, session_hash=self.request.session_hash) | |
| self.assertIn('Loaded save', result['data'][-1]) | |
| self.assertEqual(OTHER, state[app.char_id_state._id]) | |
| self.assertEqual(stored['recent_messages'], | |
| memory.normalize_history(state[app.chatbot.chatbot_state._id])) | |
| self.assertEqual(stored['relationship'], memory.load_conversation(self.session, OTHER)['relationship']) | |
| Path(gr.utils.get_upload_folder()).mkdir(parents=True, exist_ok=True) | |
| asyncio.run(run()) | |
| def test_reset_events_cancel_generations(self): | |
| self.assertGreaterEqual(len(app.chatbot.generation_events), 2) | |
| for event in app.chatbot.generation_events: | |
| self.assertIn(app.demo.fns[event['id']].name, ('respond', '_submit_fn')) | |
| config = app.demo.get_config_file() | |
| cancel_events = [d for d in config['dependencies'] if d.get('cancels')] | |
| self.assertGreaterEqual(len(cancel_events), 4) # stop, clear, switch, load | |
| if __name__ == '__main__': | |
| unittest.main() | |