test_session_reset_notify.py
1 """Tests for session auto-reset notifications. 2 3 Verifies that: 4 - _should_reset() returns a reason string ("idle" or "daily") instead of bool 5 - SessionEntry captures auto_reset_reason 6 - SessionResetPolicy.notify controls whether notifications are sent 7 - notify_exclude_platforms skips notifications for excluded platforms 8 """ 9 10 from datetime import datetime, timedelta 11 from unittest.mock import MagicMock 12 13 import pytest 14 15 from gateway.config import ( 16 GatewayConfig, 17 Platform, 18 PlatformConfig, 19 SessionResetPolicy, 20 ) 21 from gateway.session import SessionEntry, SessionSource, SessionStore 22 23 24 # --------------------------------------------------------------------------- 25 # Helpers 26 # --------------------------------------------------------------------------- 27 28 def _make_source(platform=Platform.TELEGRAM, chat_id="123", user_id="u1"): 29 return SessionSource( 30 platform=platform, 31 chat_id=chat_id, 32 user_id=user_id, 33 ) 34 35 36 def _make_store(policy=None, tmp_path=None): 37 config = GatewayConfig() 38 if policy: 39 config.default_reset_policy = policy 40 store = SessionStore(sessions_dir=tmp_path or "/tmp/test-sessions", config=config) 41 return store 42 43 44 # --------------------------------------------------------------------------- 45 # _should_reset returns reason string 46 # --------------------------------------------------------------------------- 47 48 class TestShouldResetReason: 49 def test_returns_none_when_not_expired(self, tmp_path): 50 store = _make_store( 51 SessionResetPolicy(mode="both", idle_minutes=60, at_hour=4), 52 tmp_path, 53 ) 54 entry = SessionEntry( 55 session_key="test", 56 session_id="s1", 57 created_at=datetime.now(), 58 updated_at=datetime.now(), # just updated 59 ) 60 source = _make_source() 61 assert store._should_reset(entry, source) is None 62 63 def test_returns_idle_when_idle_expired(self, tmp_path): 64 store = _make_store( 65 SessionResetPolicy(mode="idle", idle_minutes=30), 66 tmp_path, 67 ) 68 entry = SessionEntry( 69 session_key="test", 70 session_id="s1", 71 created_at=datetime.now() - timedelta(hours=2), 72 updated_at=datetime.now() - timedelta(hours=1), # 60min ago > 30min threshold 73 ) 74 source = _make_source() 75 assert store._should_reset(entry, source) == "idle" 76 77 def test_returns_daily_when_daily_boundary_crossed(self, tmp_path): 78 now = datetime.now() 79 store = _make_store( 80 SessionResetPolicy(mode="daily", at_hour=now.hour), 81 tmp_path, 82 ) 83 entry = SessionEntry( 84 session_key="test", 85 session_id="s1", 86 created_at=now - timedelta(days=2), 87 updated_at=now - timedelta(days=1), # last active yesterday 88 ) 89 source = _make_source() 90 assert store._should_reset(entry, source) == "daily" 91 92 def test_returns_none_when_mode_is_none(self, tmp_path): 93 store = _make_store( 94 SessionResetPolicy(mode="none"), 95 tmp_path, 96 ) 97 entry = SessionEntry( 98 session_key="test", 99 session_id="s1", 100 created_at=datetime.now() - timedelta(days=30), 101 updated_at=datetime.now() - timedelta(days=30), 102 ) 103 source = _make_source() 104 assert store._should_reset(entry, source) is None 105 106 107 # --------------------------------------------------------------------------- 108 # SessionEntry captures reason 109 # --------------------------------------------------------------------------- 110 111 class TestSessionEntryReason: 112 def test_auto_reset_reason_stored(self, tmp_path): 113 store = _make_store( 114 SessionResetPolicy(mode="idle", idle_minutes=1), 115 tmp_path, 116 ) 117 source = _make_source() 118 119 # Create initial session 120 entry1 = store.get_or_create_session(source) 121 assert not entry1.was_auto_reset 122 123 # Age it past the idle threshold 124 entry1.updated_at = datetime.now() - timedelta(minutes=5) 125 store._save() 126 127 # Next call should create a new session with reason 128 entry2 = store.get_or_create_session(source) 129 assert entry2.was_auto_reset is True 130 assert entry2.auto_reset_reason == "idle" 131 assert entry2.session_id != entry1.session_id 132 133 def test_reset_had_activity_false_when_no_tokens(self, tmp_path): 134 """Expired session with no tokens → reset_had_activity=False.""" 135 store = _make_store( 136 SessionResetPolicy(mode="idle", idle_minutes=1), 137 tmp_path, 138 ) 139 source = _make_source() 140 141 entry1 = store.get_or_create_session(source) 142 # No tokens used — session was idle with no conversation 143 entry1.updated_at = datetime.now() - timedelta(minutes=5) 144 store._save() 145 146 entry2 = store.get_or_create_session(source) 147 assert entry2.was_auto_reset is True 148 assert entry2.reset_had_activity is False 149 150 def test_reset_had_activity_true_when_tokens_used(self, tmp_path): 151 """Expired session with tokens → reset_had_activity=True.""" 152 store = _make_store( 153 SessionResetPolicy(mode="idle", idle_minutes=1), 154 tmp_path, 155 ) 156 source = _make_source() 157 158 entry1 = store.get_or_create_session(source) 159 # Simulate some conversation happened 160 entry1.total_tokens = 5000 161 entry1.updated_at = datetime.now() - timedelta(minutes=5) 162 store._save() 163 164 entry2 = store.get_or_create_session(source) 165 assert entry2.was_auto_reset is True 166 assert entry2.reset_had_activity is True 167 168 169 # --------------------------------------------------------------------------- 170 # SessionResetPolicy notify config 171 # --------------------------------------------------------------------------- 172 173 class TestResetPolicyNotify: 174 def test_notify_defaults_true(self): 175 policy = SessionResetPolicy() 176 assert policy.notify is True 177 178 def test_notify_exclude_defaults(self): 179 policy = SessionResetPolicy() 180 assert "api_server" in policy.notify_exclude_platforms 181 assert "webhook" in policy.notify_exclude_platforms 182 183 def test_from_dict_with_notify_false(self): 184 policy = SessionResetPolicy.from_dict({"notify": False}) 185 assert policy.notify is False 186 187 def test_from_dict_with_custom_excludes(self): 188 policy = SessionResetPolicy.from_dict({ 189 "notify_exclude_platforms": ["api_server", "webhook", "homeassistant"], 190 }) 191 assert "homeassistant" in policy.notify_exclude_platforms 192 193 def test_from_dict_preserves_defaults_on_missing_keys(self): 194 policy = SessionResetPolicy.from_dict({}) 195 assert policy.notify is True 196 assert "api_server" in policy.notify_exclude_platforms 197 198 def test_to_dict_roundtrip(self): 199 original = SessionResetPolicy( 200 mode="idle", 201 notify=False, 202 notify_exclude_platforms=("api_server",), 203 ) 204 restored = SessionResetPolicy.from_dict(original.to_dict()) 205 assert restored.notify == original.notify 206 assert restored.notify_exclude_platforms == original.notify_exclude_platforms 207 assert restored.mode == original.mode