tests/test_chat_settings.py

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