fix: honor future digest schedule changes

This commit is contained in:
VAIO73
2026-09-17 13:47:24 +02:00
parent 10a69947dc
commit 1151928b3e
2 changed files with 91 additions and 0 deletions
+33
View File
@@ -3006,6 +3006,7 @@ class NotificationManager:
def save_settings(self, settings: Dict[str, str]) -> Dict[str, Any]:
"""Save multiple notification settings at once."""
try:
previous_config = dict(self._config)
conn = sqlite3.connect(str(DB_PATH), timeout=10)
conn.execute('PRAGMA journal_mode=WAL')
conn.execute('PRAGMA busy_timeout=5000')
@@ -3098,6 +3099,38 @@ class NotificationManager:
VALUES (?, ?, ?)
''', (marker_key, 'true', now))
self._config[f'event_explicit.{event_type}'] = 'true'
# A digest time can be changed after today's digest has already
# been sent. Retaining digest_last_at in that case silently
# makes the newly selected, still-future time wait until tomorrow.
# Reset the guard only for a genuine enable/time change to a later
# time today; ordinary saves and past times remain rate-limited.
now = datetime.now()
current_minute = now.hour * 60 + now.minute
for ch_type in CHANNEL_TYPES:
enabled_key = f'{ch_type}.digest_enabled'
time_key = f'{ch_type}.digest_time'
last_key = f'{ch_type}.digest_last_at'
if self._config.get(enabled_key, 'false') != 'true':
continue
changed = (
previous_config.get(enabled_key, 'false') != 'true'
or previous_config.get(time_key, '09:00') != self._config.get(time_key, '09:00')
)
if not changed:
continue
try:
hour, minute = (int(part) for part in self._config.get(time_key, '09:00').split(':', 1))
target_minute = hour * 60 + minute
except (ValueError, AttributeError):
continue
if not (current_minute < target_minute < 24 * 60):
continue
cursor.execute('''
INSERT OR REPLACE INTO user_settings (setting_key, setting_value, updated_at)
VALUES (?, ?, ?)
''', (f'{SETTINGS_PREFIX}{last_key}', '', now.isoformat()))
self._config[last_key] = ''
conn.commit()
conn.close()
@@ -5,6 +5,7 @@ import string
import sys
import tempfile
import unittest
from datetime import datetime
from pathlib import Path
from unittest import mock
@@ -475,6 +476,63 @@ class RuntimeCatalogTests(unittest.TestCase):
self.assertEqual(loaded.get_settings()["config"]["notification_language"], "sk")
self.assertEqual(loaded.get_settings()["config"]["ai_language"], "de")
def test_future_digest_time_change_resets_only_the_digest_guard(self):
class FixedDatetime:
@classmethod
def now(cls):
return datetime(2026, 9, 17, 13, 30)
with tempfile.TemporaryDirectory() as directory:
db_path = Path(directory) / "settings.db"
conn = sqlite3.connect(db_path)
conn.execute("CREATE TABLE user_settings (setting_key TEXT PRIMARY KEY, setting_value TEXT, updated_at TEXT)")
conn.commit()
conn.close()
manager = notification_manager.NotificationManager()
manager._config = {
"telegram.digest_enabled": "true",
"telegram.digest_time": "13:20",
"telegram.digest_last_at": "2026-09-17T13:20:25",
}
with mock.patch.object(notification_manager, "DB_PATH", db_path), \
mock.patch.object(notification_manager, "datetime", FixedDatetime):
result = manager.save_settings({
"telegram.digest_enabled": "true",
"telegram.digest_time": "13:40",
})
self.assertTrue(result["success"], result)
self.assertEqual(manager._config["telegram.digest_last_at"], "")
conn = sqlite3.connect(db_path)
stored = conn.execute(
"SELECT setting_value FROM user_settings WHERE setting_key = ?",
("notification.telegram.digest_last_at",),
).fetchone()
conn.close()
self.assertEqual(stored[0], "")
manager = notification_manager.NotificationManager()
manager._config = {
"telegram.digest_enabled": "true",
"telegram.digest_time": "13:40",
"telegram.digest_last_at": "2026-09-17T13:20:25",
}
with tempfile.TemporaryDirectory() as directory:
db_path = Path(directory) / "settings.db"
conn = sqlite3.connect(db_path)
conn.execute("CREATE TABLE user_settings (setting_key TEXT PRIMARY KEY, setting_value TEXT, updated_at TEXT)")
conn.commit()
conn.close()
with mock.patch.object(notification_manager, "DB_PATH", db_path), \
mock.patch.object(notification_manager, "datetime", FixedDatetime):
result = manager.save_settings({
"telegram.digest_enabled": "true",
"telegram.digest_time": "13:40",
})
self.assertTrue(result["success"], result)
self.assertEqual(manager._config["telegram.digest_last_at"], "2026-09-17T13:20:25")
def test_ai_language_is_independent_from_runtime_notification_language(self):
manager = notification_manager.NotificationManager()
manager._config = {