gooneyryan/hermes-webui
1
0
Fork 0
mirror of https://github.com/nesquena/hermes-webui.git synced 2026-05-25 19:20:16 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
0774235987bae61ba04a44dcaad22e71ca33ec70
hermes-webui/tests/test_1058_adaptive_title_refresh.py
T

417 lines
17 KiB
Python
Raw Blame History

"""Tests for adaptive session title refresh helpers (PR #1058).
Covers all five new functions added to api/streaming.py:
- _count_exchanges
- _latest_exchange_snippets
- _get_title_refresh_interval
- _run_background_title_refresh
- _maybe_schedule_title_refresh
"""
import sys
import os
import threading
import types
import unittest
from unittest.mock import MagicMock, patch
import pytest
# Ensure the project root is on sys.path
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
from api.streaming import (
_count_exchanges,
_latest_exchange_snippets,
_get_title_refresh_interval,
_run_background_title_refresh,
_maybe_schedule_title_refresh,
)
@pytest.fixture(autouse=True)
def _restore_auth_sessions():
"""Snapshot and restore api.auth._sessions around each test.
Importing api.streaming can trigger api.config.load_settings() which may
call into api.auth and create a real session token. Without this fixture,
that stale token leaks into test_auth_session_persistence.py tests (which
assume _sessions starts empty) when our file runs first alphabetically.
"""
import api.auth as _auth
snapshot = dict(_auth._sessions)
yield
_auth._sessions.clear()
_auth._sessions.update(snapshot)
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _user_msg(text):
return {'role': 'user', 'content': text}
def _asst_msg(text, tool_calls=None):
msg = {'role': 'assistant', 'content': text}
if tool_calls is not None:
msg['tool_calls'] = tool_calls
return msg
def _tool_only_asst():
"""Assistant message that only has tool_calls, no real text."""
return {'role': 'assistant', 'content': '', 'tool_calls': [{'id': 't1', 'type': 'function'}]}
def _make_session(title='My Title', llm_title_generated=True, messages=None, session_id='sid1'):
s = MagicMock()
s.title = title
s.llm_title_generated = llm_title_generated
s.messages = messages or []
s.session_id = session_id
s.save = MagicMock()
return s
# ---------------------------------------------------------------------------
# _count_exchanges
# ---------------------------------------------------------------------------
class TestCountExchanges:
def test_empty_messages_returns_zero(self):
assert _count_exchanges([]) == 0
def test_none_messages_returns_zero(self):
assert _count_exchanges(None) == 0
def test_counts_only_user_messages(self):
msgs = [_user_msg('hello'), _asst_msg('hi'), _user_msg('world')]
assert _count_exchanges(msgs) == 2
def test_skips_empty_user_messages(self):
msgs = [_user_msg(''), _user_msg(' '), _user_msg('real question')]
assert _count_exchanges(msgs) == 1
def test_counts_list_content_user_messages(self):
msgs = [
{'role': 'user', 'content': [{'type': 'text', 'text': 'list question'}]},
]
assert _count_exchanges(msgs) == 1
def test_skips_empty_list_content(self):
msgs = [
{'role': 'user', 'content': [{'type': 'text', 'text': ' '}]},
]
assert _count_exchanges(msgs) == 0
def test_ignores_non_user_roles(self):
msgs = [_asst_msg('response'), {'role': 'system', 'content': 'system prompt'}]
assert _count_exchanges(msgs) == 0
def test_ignores_non_dict_entries(self):
msgs = ['not a dict', _user_msg('real'), None]
assert _count_exchanges(msgs) == 1
def test_five_exchanges(self):
msgs = []
for i in range(5):
msgs.append(_user_msg(f'question {i}'))
msgs.append(_asst_msg(f'answer {i}'))
assert _count_exchanges(msgs) == 5
# ---------------------------------------------------------------------------
# _latest_exchange_snippets
# ---------------------------------------------------------------------------
class TestLatestExchangeSnippets:
def test_empty_returns_empty_strings(self):
u, a = _latest_exchange_snippets([])
assert u == '' and a == ''
def test_none_returns_empty_strings(self):
u, a = _latest_exchange_snippets(None)
assert u == '' and a == ''
def test_basic_pair(self):
msgs = [_user_msg('first q'), _asst_msg('first a'),
_user_msg('second q'), _asst_msg('second a')]
u, a = _latest_exchange_snippets(msgs)
assert u == 'second q'
assert a == 'second a'
def test_returns_latest_not_first(self):
msgs = [_user_msg('old q'), _asst_msg('old a'),
_user_msg('new q'), _asst_msg('new a')]
u, a = _latest_exchange_snippets(msgs)
assert u == 'new q'
def test_skips_tool_call_only_assistant(self):
"""An assistant msg with tool_calls and no real text should be skipped."""
msgs = [_user_msg('q'), _asst_msg('real answer'),
_user_msg('q2'), _tool_only_asst()]
u, a = _latest_exchange_snippets(msgs)
# _tool_only_asst should be skipped; fall back to previous real assistant
assert a == 'real answer'
assert u == 'q2'
def test_truncates_long_content(self):
long_text = 'x' * 600
msgs = [_user_msg(long_text), _asst_msg(long_text)]
u, a = _latest_exchange_snippets(msgs)
assert len(u) == 500
assert len(a) == 500
def test_no_assistant_message(self):
msgs = [_user_msg('q')]
u, a = _latest_exchange_snippets(msgs)
assert u == 'q'
assert a == ''
def test_no_user_message(self):
msgs = [_asst_msg('a')]
u, a = _latest_exchange_snippets(msgs)
assert u == ''
assert a == 'a'
def test_ignores_non_dict_entries(self):
msgs = ['noise', _user_msg('q'), None, _asst_msg('a')]
u, a = _latest_exchange_snippets(msgs)
assert u == 'q'
assert a == 'a'
# ---------------------------------------------------------------------------
# _get_title_refresh_interval
# ---------------------------------------------------------------------------
class TestGetTitleRefreshInterval:
def test_returns_int_for_valid_setting(self):
# _get_title_refresh_interval does a local import: `from api.config import load_settings`
# so patch the source module, not api.streaming
with patch('api.config.load_settings', return_value={'auto_title_refresh_every': '5'}):
assert _get_title_refresh_interval() == 5
def test_returns_zero_for_off_setting(self):
with patch('api.config.load_settings', return_value={'auto_title_refresh_every': '0'}):
assert _get_title_refresh_interval() == 0
def test_returns_zero_when_key_absent(self):
with patch('api.config.load_settings', return_value={}):
assert _get_title_refresh_interval() == 0
def test_returns_zero_on_exception(self):
with patch('api.config.load_settings', side_effect=Exception('boom')):
assert _get_title_refresh_interval() == 0
def test_valid_values_10_and_20(self):
for val in ('10', '20'):
with patch('api.config.load_settings', return_value={'auto_title_refresh_every': val}):
assert _get_title_refresh_interval() == int(val)
# ---------------------------------------------------------------------------
# _run_background_title_refresh
# ---------------------------------------------------------------------------
class TestRunBackgroundTitleRefresh:
def _make_put_event(self):
events = []
def put(name, data):
events.append((name, data))
return put, events
def _make_session_obj(self, title='Old Title'):
s = MagicMock()
s.title = title
s.llm_title_generated = True
s.save = MagicMock()
return s
def test_skips_when_title_changed_before_call(self):
"""If the title has changed (manual rename) since the refresh was scheduled, skip."""
put, events = self._make_put_event()
with patch('api.streaming.get_session') as mock_get, \
patch('api.streaming.SESSIONS', {}), \
patch('api.streaming.LOCK', threading.Lock()):
s = self._make_session_obj(title='Different Title')
mock_get.return_value = s
_run_background_title_refresh(
'sid', 'user', 'asst', 'Old Title', put, agent=None
)
# No 'title' event should have been emitted
assert not any(name == 'title' for name, _ in events)
def test_skips_if_session_not_found(self):
put, events = self._make_put_event()
with patch('api.streaming.get_session', side_effect=KeyError('not found')):
_run_background_title_refresh('sid', 'u', 'a', 'title', put)
assert events == []
def test_skips_when_title_is_untitled(self):
put, events = self._make_put_event()
with patch('api.streaming.get_session') as mock_get:
s = self._make_session_obj(title='Untitled')
mock_get.return_value = s
_run_background_title_refresh('sid', 'u', 'a', 'Untitled', put)
assert not any(name == 'title' for name, _ in events)
def test_skips_same_title(self):
"""If the LLM generates a title identical to the current one, no event is emitted."""
put, events = self._make_put_event()
with patch('api.streaming.get_session') as mock_get, \
patch('api.streaming._aux_title_configured', return_value=True), \
patch('api.streaming._generate_llm_session_title_via_aux',
return_value=('Old Title', 'llm_ok', 'raw')), \
patch('api.streaming.SESSIONS', {}), \
patch('api.streaming.LOCK', threading.Lock()):
s = self._make_session_obj(title='Old Title')
mock_get.return_value = s
_run_background_title_refresh('sid', 'u', 'a', 'Old Title', put)
assert not any(name == 'title' for name, _ in events)
def test_emits_title_event_on_new_title(self):
put, events = self._make_put_event()
s = self._make_session_obj(title='Old Title')
# Use a real dict for SESSIONS so .get() works, pre-populated with our session
fake_sessions = {'sid': s}
with patch('api.streaming.get_session', return_value=s), \
patch('api.streaming._aux_title_configured', return_value=True), \
patch('api.streaming._generate_llm_session_title_via_aux',
return_value=('New Refreshed Title', 'llm_ok', 'raw')), \
patch('api.streaming.SESSIONS', fake_sessions), \
patch('api.streaming.LOCK', threading.Lock()):
_run_background_title_refresh('sid', 'u', 'a', 'Old Title', put)
title_events = [(n, d) for n, d in events if n == 'title']
assert len(title_events) == 1
assert title_events[0][1]['title'] == 'New Refreshed Title'
def test_saves_refreshed_title_outside_global_lock(self):
"""Refreshing an existing title must not call Session.save() while holding LOCK."""
class TrackingLock:
def __init__(self):
self.held = False
def __enter__(self):
assert not self.held
self.held = True
return self
def __exit__(self, exc_type, exc, tb):
self.held = False
put, events = self._make_put_event()
lock = TrackingLock()
s = self._make_session_obj(title='Old Title')
def save(*args, **kwargs):
assert not lock.held, "Session.save() must run outside api.models.LOCK"
s.save = save
fake_sessions = {'sid': s}
with patch('api.streaming.get_session', return_value=s), \
patch('api.streaming._aux_title_configured', return_value=True), \
patch('api.streaming._generate_llm_session_title_via_aux',
return_value=('New Refreshed Title', 'llm_ok', 'raw')), \
patch('api.streaming.SESSIONS', fake_sessions), \
patch('api.streaming.LOCK', lock):
_run_background_title_refresh('sid', 'u', 'a', 'Old Title', put)
title_events = [(n, d) for n, d in events if n == 'title']
assert len(title_events) == 1
assert title_events[0][1]['title'] == 'New Refreshed Title'
def test_exceptions_are_silently_swallowed(self):
"""Any unexpected error inside must not propagate — it's a background daemon."""
put, events = self._make_put_event()
with patch('api.streaming.get_session', side_effect=RuntimeError('oops')):
# Should not raise
_run_background_title_refresh('sid', 'u', 'a', 'title', put)
assert events == []
# ---------------------------------------------------------------------------
# _maybe_schedule_title_refresh
# ---------------------------------------------------------------------------
class TestMaybeScheduleTitleRefresh:
def _noop_put(self, name, data):
pass
def test_does_nothing_when_disabled(self):
with patch('api.streaming._get_title_refresh_interval', return_value=0):
spawned = []
with patch('threading.Thread', side_effect=lambda **kw: spawned.append(kw) or MagicMock()):
session = _make_session(messages=[_user_msg('q'), _asst_msg('a')] * 5)
_maybe_schedule_title_refresh(session, self._noop_put, None)
assert spawned == []
def test_does_nothing_when_title_is_empty(self):
with patch('api.streaming._get_title_refresh_interval', return_value=5):
spawned = []
with patch('threading.Thread', side_effect=lambda **kw: spawned.append(kw) or MagicMock()):
session = _make_session(title='', messages=[_user_msg('q'), _asst_msg('a')] * 5)
_maybe_schedule_title_refresh(session, self._noop_put, None)
assert spawned == []
def test_does_nothing_for_untitled(self):
with patch('api.streaming._get_title_refresh_interval', return_value=5):
spawned = []
with patch('threading.Thread', side_effect=lambda **kw: spawned.append(kw) or MagicMock()):
session = _make_session(title='Untitled', messages=[_user_msg('q'), _asst_msg('a')] * 5)
_maybe_schedule_title_refresh(session, self._noop_put, None)
assert spawned == []
def test_does_nothing_when_title_not_llm_generated(self):
with patch('api.streaming._get_title_refresh_interval', return_value=5):
spawned = []
with patch('threading.Thread', side_effect=lambda **kw: spawned.append(kw) or MagicMock()):
session = _make_session(llm_title_generated=False,
messages=[_user_msg('q'), _asst_msg('a')] * 5)
_maybe_schedule_title_refresh(session, self._noop_put, None)
assert spawned == []
def test_does_nothing_when_exchange_count_not_at_interval(self):
"""Refresh only fires when exchange_count % interval == 0 (and > 0)."""
with patch('api.streaming._get_title_refresh_interval', return_value=5):
spawned = []
with patch('threading.Thread', side_effect=lambda **kw: spawned.append(kw) or MagicMock()):
# 4 exchanges — not a multiple of 5
session = _make_session(messages=[_user_msg('q'), _asst_msg('a')] * 4)
_maybe_schedule_title_refresh(session, self._noop_put, None)
assert spawned == []
def test_spawns_thread_at_exact_interval(self):
"""Refresh fires when exchange_count == refresh_interval."""
with patch('api.streaming._get_title_refresh_interval', return_value=5):
spawned = []
with patch('threading.Thread') as mock_thread_cls:
mock_thread = MagicMock()
mock_thread_cls.return_value = mock_thread
# 5 user messages = 5 exchanges
session = _make_session(messages=[_user_msg('q'), _asst_msg('a')] * 5)
_maybe_schedule_title_refresh(session, self._noop_put, None)
assert mock_thread_cls.called
assert mock_thread.start.called
def test_spawns_thread_at_multiple_of_interval(self):
"""Refresh fires at 10 exchanges when interval is 5."""
with patch('api.streaming._get_title_refresh_interval', return_value=5):
with patch('threading.Thread') as mock_thread_cls:
mock_thread = MagicMock()
mock_thread_cls.return_value = mock_thread
# 10 exchanges
session = _make_session(messages=[_user_msg('q'), _asst_msg('a')] * 10)
_maybe_schedule_title_refresh(session, self._noop_put, None)
assert mock_thread_cls.called
def test_does_nothing_when_no_exchange_content(self):
"""Even at interval, if both snippets are empty, don't spawn."""
with patch('api.streaming._get_title_refresh_interval', return_value=5), \
patch('api.streaming._latest_exchange_snippets', return_value=('', '')):
spawned = []
with patch('threading.Thread', side_effect=lambda **kw: spawned.append(kw) or MagicMock()):
session = _make_session(messages=[_user_msg('q'), _asst_msg('a')] * 5)
_maybe_schedule_title_refresh(session, self._noop_put, None)
assert spawned == []
Reference in New Issue View Git Blame Copy Permalink