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