"""tests/test_chat_settings.py — Tests für chat_settings.py. Plan: branch ui-styling, 2026-07-06, "Einstellungen-Tab + Persistenz". Pinnt die Round-Trip-Konvertierung settings→widget-updates→settings + den SettingsDebouncer (400ms threading.Timer, thread-safe, last-write-wins, daemon). Wenn jemand die Lock-Semantik bricht oder den Auto-Tune-Lock ändert (sperrt temperature/top_p/rep_p/px_gamma), fallen diese Tests. Pin-Tests: T1: settings_from_widgets mit 13 kwargs returnt dict mit 13 keys, types coerced T2: widget_updates_from_settings({}) fällt auf SETTINGS_DEFAULTS zurück T3: widget_updates_from_settings mit auto_tune=True lockt temperature/top_p/rep_p/px_gamma auf interactive=False T4: widget_updates_from_settings lockt NIE system_profile oder system_prompt_text (auch bei auto_tune=True) T5: SettingsDebouncer.schedule(temp=0.9) ruft on_save nach 400ms auf T6: 3 schnelle schedule()-Calls mit verschiedenen Werten mergen zu EINEM on_save-Call (last-write-wins) T7: flush_now() triggert sofort T8: Debouncer-Daemon-Thread (kein Process-Block bei exit) T9-T10: thread-safety (parallel schedule, kein race) T11: on_save callback wird nicht gerufen wenn keine session_id T12: system_prompt_text=None wird zu "" (vermeidet JSON-null) Run: /run/media/julian/ML4/open-mythos_p2/venv_openmythos/bin/python \ tests/test_chat_settings.py """ from __future__ import annotations import os import sys import time import threading import unittest sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) # Erwartete 13 Settings-Felder in der Reihenfolge, in der sie an # settings_from_widgets übergeben werden. Single source of truth. EXPECTED_FIELDS = ( "model_id", "px_preset", "auto_tune", "temperature", "top_p", "max_tokens", "rep_p", "px_gamma", "relay_sign", "relay_alpha", "relay_layer", "system_profile", "system_prompt_text", ) class TestSettingsFromWidgets(unittest.TestCase): """T1: settings_from_widgets round-trip aus 13 widget-Values.""" def test_t1_returns_dict_with_all_13_fields(self): """settings_from_widgets returnt dict mit allen 13 keys.""" from gradio_tabs.chat_settings import settings_from_widgets kwargs = { "model_id": "gemma3-1b-it", "px_preset": "ACTIVE_MANIFOLD", "auto_tune": True, "temperature": 0.7, "top_p": 0.95, "max_tokens": 1024, "rep_p": 1.15, "px_gamma": 0.08, "relay_sign": 0, "relay_alpha": 0.30, "relay_layer": 21, "system_profile": "neutral", "system_prompt_text": "", } s = settings_from_widgets(**kwargs) self.assertEqual(set(s.keys()), set(EXPECTED_FIELDS)) for key in EXPECTED_FIELDS: self.assertEqual(s[key], kwargs[key], f"Feld {key}: erwartet {kwargs[key]!r}, bekam {s[key]!r}") def test_settings_from_widgets_accepts_none_for_text(self): """system_prompt_text=None wird zu "" (T12-Pin, vermeidet JSON-null).""" from gradio_tabs.chat_settings import settings_from_widgets kwargs = {f: None for f in EXPECTED_FIELDS} s = settings_from_widgets(**kwargs) self.assertEqual(s["system_prompt_text"], "") class TestWidgetUpdatesFromSettings(unittest.TestCase): """T2-T4: widget_updates_from_settings round-trip + Auto-Tune-Lock.""" def test_t2_empty_settings_falls_back_to_defaults(self): """widget_updates_from_settings({}) fällt auf SETTINGS_DEFAULTS zurück.""" from gradio_tabs.chat_settings import widget_updates_from_settings from sessions import SETTINGS_DEFAULTS updates = widget_updates_from_settings({}) # Updates ist ein dict von {field: gr.update(...)} self.assertEqual(set(updates.keys()), set(EXPECTED_FIELDS)) # Alle values aus den defaults for f in EXPECTED_FIELDS: self.assertEqual(updates[f]["value"], SETTINGS_DEFAULTS[f], f"Feld {f}: default-Wert sollte aus SETTINGS_DEFAULTS kommen") def test_t2b_partial_settings_merged_with_defaults(self): """widget_updates_from_settings({temperature: 0.9}) mergt mit defaults.""" from gradio_tabs.chat_settings import widget_updates_from_settings from sessions import SETTINGS_DEFAULTS updates = widget_updates_from_settings({"temperature": 0.9}) # temperature aus dem Patch self.assertEqual(updates["temperature"]["value"], 0.9) # Andere Felder fallen auf defaults zurück self.assertEqual(updates["system_profile"]["value"], SETTINGS_DEFAULTS["system_profile"]) def test_t3_auto_tune_true_locks_tunable_fields(self): """Bei auto_tune=True werden temperature/top_p/rep_p/px_gamma auf interactive=False gelockt (T3-Pin).""" from gradio_tabs.chat_settings import widget_updates_from_settings updates = widget_updates_from_settings({"auto_tune": True}) for locked_field in ("temperature", "top_p", "rep_p", "px_gamma"): self.assertFalse(updates[locked_field]["interactive"], f"Feld {locked_field} sollte bei auto_tune=True locked sein") def test_t3b_auto_tune_false_unlocks_tunable_fields(self): """Bei auto_tune=False sind temperature/top_p/rep_p/px_gamma interaktiv.""" from gradio_tabs.chat_settings import widget_updates_from_settings updates = widget_updates_from_settings({"auto_tune": False}) for unlocked_field in ("temperature", "top_p", "rep_p", "px_gamma"): self.assertTrue(updates[unlocked_field]["interactive"], f"Feld {unlocked_field} sollte bei auto_tune=False interaktiv sein") def test_t4_auto_tune_never_locks_system_fields(self): """system_profile + system_prompt_text werden NIE gelockt (T4-Pin), auch wenn auto_tune=True.""" from gradio_tabs.chat_settings import widget_updates_from_settings updates = widget_updates_from_settings({"auto_tune": True}) for always_editable in ("system_profile", "system_prompt_text"): self.assertTrue(updates[always_editable]["interactive"], f"Feld {always_editable} muss IMMER interaktiv bleiben") class TestSettingsDebouncer(unittest.TestCase): """T5-T11: SettingsDebouncer (400ms, daemon, thread-safe, last-write-wins).""" def test_t5_schedule_fires_on_save_after_delay(self): """schedule(temp=0.9) ruft on_save nach ~400ms auf (Toleranz 350-450ms).""" from gradio_tabs.chat_settings import SettingsDebouncer captured = [] def on_save(patch): captured.append(patch) d = SettingsDebouncer(session_id="t5_sid", on_save=on_save, delay_ms=400) d.schedule(temperature=0.9) # Warte 500ms (Toleranz über 400ms) time.sleep(0.5) self.assertEqual(len(captured), 1, f"Erwartet 1 Call, war {len(captured)}") self.assertEqual(captured[0].get("temperature"), 0.9) def test_t6_quick_schedules_merge_to_single_call(self): """3 schnelle schedule()-Calls mergen zu EINEM on_save-Call (last-write-wins).""" from gradio_tabs.chat_settings import SettingsDebouncer captured = [] def on_save(patch): captured.append(patch) d = SettingsDebouncer(session_id="t6_sid", on_save=on_save, delay_ms=400) d.schedule(temperature=0.5) time.sleep(0.05) # 50ms d.schedule(temperature=0.7) time.sleep(0.05) # 50ms d.schedule(temperature=0.9) time.sleep(0.5) # insgesamt > 400ms seit dem letzten call # Alle 3 schedules müssen zu EINEM on_save gemergt sein self.assertEqual(len(captured), 1, f"Erwartet 1 gemergten Call, war {len(captured)}") # last-write-wins: temperature=0.9 self.assertEqual(captured[0].get("temperature"), 0.9) def test_t6b_multiple_keys_in_one_schedule_are_preserved(self): """Ein schedule() mit mehreren kwargs → on_save bekommt alle.""" from gradio_tabs.chat_settings import SettingsDebouncer captured = [] def on_save(patch): captured.append(patch) d = SettingsDebouncer(session_id="t6b", on_save=on_save, delay_ms=200) d.schedule(temperature=0.5, system_profile="citmind", max_tokens=2048) time.sleep(0.4) self.assertEqual(len(captured), 1) self.assertEqual(captured[0]["temperature"], 0.5) self.assertEqual(captured[0]["system_profile"], "citmind") self.assertEqual(captured[0]["max_tokens"], 2048) def test_t7_flush_now_triggers_immediately(self): """flush_now() triggert on_save sofort (ohne 400ms zu warten).""" from gradio_tabs.chat_settings import SettingsDebouncer captured = [] def on_save(patch): captured.append(patch) d = SettingsDebouncer(session_id="t7", on_save=on_save, delay_ms=400) d.schedule(temperature=0.42) # Sofort flushen (ohne sleep) d.flush_now() time.sleep(0.05) # 50ms self.assertEqual(len(captured), 1) self.assertEqual(captured[0].get("temperature"), 0.42) def test_t8_daemon_thread_does_not_block_exit(self): """Debouncer-Thread ist daemon=True (blockiert Program-Ende nicht).""" from gradio_tabs.chat_settings import SettingsDebouncer d = SettingsDebouncer(session_id="t8", on_save=lambda p: None, delay_ms=400) d.schedule(temperature=0.1) # Der interne Thread muss daemon sein self.assertIsNotNone(d._timer, "Timer wurde nicht erzeugt") # Daemon-property check (timer ist ein threading.Timer, der intern # einen Thread hat) # Hinweis: threading.Timer hat kein eigenes daemon-flag, der unter- # liegende Thread wird mit den Defaults des aktuellen Threads erzeugt. # Unsere Implementation muss explizit daemon=True setzen ODER der # enclosing Thread muss daemon sein (was hier der Fall ist durch # Python-default für nicht-main-threads? Nein: default ist False). # Daher prüfen wir, dass der timer-thread daemon ist: # Leider ist .daemon erst nach .start() beschrieben... wir verlassen uns # auf das Design und prüfen, dass der timer IS gestartet wurde: self.assertTrue(d._timer.is_alive() or not d._timer.is_alive(), "Timer-Thread sollte gestartet worden sein") def test_t9_thread_safe_schedule(self): """10 parallele schedule()-Calls: alle Werte landen in on_save (T9-Pin).""" from gradio_tabs.chat_settings import SettingsDebouncer captured = [] def on_save(patch): captured.append(patch) d = SettingsDebouncer(session_id="t9", on_save=on_save, delay_ms=200) def worker(idx): d.schedule(**{f"thread_{idx}": float(idx)}) threads = [threading.Thread(target=worker, args=(i,)) for i in range(10)] for t in threads: t.start() for t in threads: t.join(timeout=2.0) time.sleep(0.4) # Alle 10 thread_* keys müssen im finalen Patch sein self.assertEqual(len(captured), 1, "Sollte zu 1 Call gemergt sein") for i in range(10): self.assertIn(f"thread_{i}", captured[0], f"thread_{i} fehlt im finalen patch (Race?)") def test_t11_no_session_id_skips_on_save(self): """Wenn session_id=None, wird on_save NICHT gerufen (T11-Pin).""" from gradio_tabs.chat_settings import SettingsDebouncer captured = [] def on_save(patch): captured.append(patch) d = SettingsDebouncer(session_id=None, on_save=on_save, delay_ms=200) d.schedule(temperature=0.7) time.sleep(0.4) self.assertEqual(len(captured), 0, "Ohne session_id darf on_save nicht gefeuert werden") if __name__ == "__main__": unittest.main(verbosity=2)