import json import re import sqlite3 import string import sys import tempfile import unittest from pathlib import Path from unittest import mock SCRIPTS_DIR = Path(__file__).resolve().parents[1] APPIMAGE_DIR = SCRIPTS_DIR.parent if str(SCRIPTS_DIR) not in sys.path: sys.path.insert(0, str(SCRIPTS_DIR)) import notification_manager import notification_templates import notification_channels def _placeholders(value): return { field_name for _literal, field_name, _format_spec, _conversion in string.Formatter().parse(value) if field_name } class RuntimeCatalogTests(unittest.TestCase): RUNTIME_LANGUAGES = ("en", "de", "es", "fr", "it", "pt", "sk", "sv") @classmethod def setUpClass(cls): cls.catalogs = {} for language in cls.RUNTIME_LANGUAGES: path = APPIMAGE_DIR / "messages" / language / "common.json" cls.catalogs[language] = json.loads(path.read_text(encoding="utf-8"))["runtime"]["notifications"] def test_runtime_catalog_covers_every_template_dynamically(self): expected = set(notification_templates.TEMPLATES) for language, catalog in self.catalogs.items(): templates = catalog["templates"] self.assertEqual(set(templates), expected, language) for event_type, source in notification_templates.TEMPLATES.items(): self.assertEqual(set(templates[event_type]), {"title", "body", "label"}) for field in ("title", "body", "label"): self.assertIsInstance(templates[event_type][field], str) self.assertTrue(templates[event_type][field]) if field in source: self.assertEqual( _placeholders(templates[event_type][field]), _placeholders(source[field]), f"{language}:{event_type}:{field}", ) def test_runtime_catalog_keys_and_placeholders_match(self): def flatten(value, prefix=""): result = {} for key, child in value.items(): dotted = f"{prefix}.{key}" if prefix else key if isinstance(child, dict): result.update(flatten(child, dotted)) else: result[dotted] = child return result en = flatten(self.catalogs["en"]) for language, catalog in self.catalogs.items(): translated = flatten(catalog) self.assertEqual(set(translated), set(en), language) for key in en: self.assertEqual(_placeholders(translated[key]), _placeholders(en[key]), f"{language}:{key}") def test_notification_language_ui_keys_exist_in_both_catalogs(self): required = { "notificationLanguage", "selectNotificationLanguage", "notificationLanguageHint", } for language in self.RUNTIME_LANGUAGES: path = APPIMAGE_DIR / "messages" / language / "common.json" common = json.loads(path.read_text(encoding="utf-8")) ui = common["settings"]["notifications"]["ui"] self.assertTrue(required.issubset(ui), language) for key in required: self.assertIsInstance(ui[key], str) self.assertTrue(ui[key].strip(), f"{language}:{key}") def test_slovak_catalog_preserves_placeholders_and_translates_static_text(self): en = self.catalogs["en"]["templates"] sk = self.catalogs["sk"]["templates"] for event_type, source in notification_templates.TEMPLATES.items(): for field in ("title", "body", "label"): static_text = source.get(field, "") for placeholder in _placeholders(static_text): static_text = static_text.replace("{" + placeholder + "}", "") if any(ch.isalpha() for ch in static_text): self.assertNotEqual(sk[event_type][field], en[event_type][field], f"{event_type}:{field}") def test_every_template_renders_in_slovak_and_preserves_dynamic_values(self): values = { name: f"DYNAMIC_{name.upper()}" for template in notification_templates.TEMPLATES.values() for field in ("title", "body") for name in _placeholders(template.get(field, "")) } values.update({"hostname": "HOST-ŽILINA", "severity": "WARNING"}) with mock.patch.object(notification_templates, "_get_hostname", return_value="HOST-ŽILINA"): for event_type in notification_templates.TEMPLATES: rendered = notification_templates.render_template(event_type, values, language="sk") combined = rendered["title"] + "\n" + rendered["body"] if notification_templates.TEMPLATES[event_type].get("formatter"): continue for name in _placeholders( notification_templates.TEMPLATES[event_type]["title"] + notification_templates.TEMPLATES[event_type]["body"] ): if name not in {"entity_suffix", "title_or_default"}: self.assertIn(str(values[name]), combined, f"{event_type}:{name}") def test_lxc_update_result_and_details_render_in_slovak(self): data = { "hostname": "homelab", "ct_name": "iventoy", "vmid": "105", "result": "succeeded", "details": "Source: Manual", "lxc_update": { "status": "success", "source": "manual", "targets": ["app:iventoy"], "labels": ["iVentoy"], "duration": "16s", "before": { "apps": {"iventoy": {"name": "iVentoy", "installed_version": "1.0.42"}}, }, "after": { "apps": {"iventoy": {"name": "iVentoy", "installed_version": "1.0.43"}}, }, "verification_pending": False, "verification_errors": [], "reboot_required": False, }, } rendered = notification_templates.render_template("lxc_update_applied", data, language="sk") self.assertEqual(rendered["title"], "homelab: LXC iventoy (105) aktualizácia úspešne dokončená") self.assertIn("Zdroj: Manuálne", rendered["body"]) self.assertIn("Ciele: iVentoy", rendered["body"]) self.assertIn("Aplikácie: iVentoy: 1.0.42 → 1.0.43", rendered["body"]) self.assertIn("Vyžaduje sa reštart: nie", rendered["body"]) self.assertIn("Trvanie: 16s", rendered["body"]) self.assertNotIn("Source:", rendered["body"]) self.assertNotIn("succeeded", rendered["title"]) _title, enriched_body = notification_templates.enrich_with_emojis( "lxc_update_applied", rendered["title"], rendered["body"], {**data, "_notification_language": "sk", "severity": "INFO"}, ) self.assertIn("🧭 Zdroj: Manuálne", enriched_body) self.assertIn("🎯 Ciele: iVentoy", enriched_body) self.assertIn("🧩 Aplikácie: iVentoy: 1.0.42 → 1.0.43", enriched_body) self.assertIn("🔄 Vyžaduje sa reštart: nie", enriched_body) self.assertIn("⏱️ Trvanie: 16s", enriched_body) for language in self.RUNTIME_LANGUAGES: rendered_locale = notification_templates.render_template( "lxc_update_applied", data, language=language, ) _title, body_locale = notification_templates.enrich_with_emojis( "lxc_update_applied", rendered_locale["title"], rendered_locale["body"], {**data, "_notification_language": language, "severity": "INFO"}, ) self.assertIn("🧭", body_locale, language) self.assertIn("⏱️", body_locale, language) def test_update_summary_body_icons_follow_the_selected_language(self): data = { "hostname": "pve01", "total_count": "2", "security_count": "0", "pve_count": "1", "kernel_count": "0", "important_list": "none", "severity": "INFO", "_notification_language": "sk", } rendered = notification_templates.render_template("update_summary", data, language="sk") _title, enriched_body = notification_templates.enrich_with_emojis( "update_summary", rendered["title"], rendered["body"], data, ) self.assertIn("📦 Aktualizácie spolu: 2", enriched_body) self.assertIn("🛡️ Bezpečnostné aktualizácie: 0", enriched_body) self.assertIn("⚙️ Aktualizácie jadra: 0", enriched_body) self.assertIn("📋 Dôležité balíky:", enriched_body) self.assertIn("žiadne", enriched_body) self.assertNotIn("\nnone", enriched_body) for language in self.RUNTIME_LANGUAGES: locale_data = {**data, "_notification_language": language} rendered_locale = notification_templates.render_template( "update_summary", locale_data, language=language, ) _title, body_locale = notification_templates.enrich_with_emojis( "update_summary", rendered_locale["title"], rendered_locale["body"], locale_data, ) total_label = notification_templates._localized_template_labels( "update_summary", language, )["total_count"][0] self.assertIn(f"📦 {total_label}: 2", body_locale, language) def test_docker_update_body_icons_preserve_localized_container_label(self): data = { "hostname": "pve01", "vmid": "210", "ct_name": "repopulse-labs-test", "count": "1", "details": "• Docker Engine: 29.8.0 → 29.8.1", "severity": "INFO", "_notification_language": "sk", } rendered = notification_templates.render_template( "docker_stack_update_available", data, language="sk", ) _title, enriched_body = notification_templates.enrich_with_emojis( "docker_stack_update_available", rendered["title"], rendered["body"], data, ) self.assertIn("📦 Kontajner repopulse-labs-test (CT 210) má 1 aktualizácií Docker:", enriched_body) self.assertIn("🐳 • Docker Engine: 29.8.0 → 29.8.1", enriched_body) def test_missing_slovak_key_falls_back_to_english(self): with tempfile.TemporaryDirectory() as directory: root = Path(directory) (root / "en").mkdir() (root / "sk").mkdir() (root / "en" / "common.json").write_text( json.dumps({"runtime": {"notifications": {"fallback": {"unknownTitle": "{hostname}: {event_type}"}}}}), encoding="utf-8", ) (root / "sk" / "common.json").write_text( json.dumps({"runtime": {"notifications": {}}}), encoding="utf-8" ) with mock.patch.object(notification_templates, "RUNTIME_CATALOG_DIR", root): notification_templates._load_runtime_catalog.cache_clear() self.assertEqual( notification_templates.runtime_message( "fallback.unknownTitle", "sk", hostname="pve01", event_type="vendor_event" ), "pve01: vendor_event", ) notification_templates._load_runtime_catalog.cache_clear() def test_special_formatters_digest_and_test_message_are_slovak(self): startup = notification_templates.render_template( "system_startup", {"hostname": "pve01", "has_issues": False, "vms_started": [{"name": "alpha", "vmid": 100}]}, language="sk", ) self.assertIn("Spustenie systému", startup["title"]) self.assertIn("Všetky systémy sú funkčné", startup["body"]) self.assertIn("alpha", startup["body"]) app = notification_templates.render_template( "app_update_available", {"hostname": "pve01", "app_name": "Redis", "vmid": 115, "ct_name": "cache", "installed": "7.0", "latest": "8.1"}, language="sk", ) self.assertIn("dostupná aktualizácia", app["title"]) self.assertIn("Redis", app["body"]) self.assertIn("7.0 → 8.1", app["body"]) backup = notification_templates.render_template( "backup_complete", { "hostname": "pve01", "storage": "pbs-main", "vmname": "alpha", "vmid": "100", "pve_title": "Backup job finished", "pve_message": ( "INFO: Starting Backup of VM 100 (qemu)\n" "INFO: VM Name: alpha\n" "INFO: transferred 1.5 GiB in 10 seconds\n" "INFO: Finished Backup of VM 100 (00:00:10)" ), }, language="sk", ) self.assertIn("Záloha dokončená", backup["title"]) self.assertNotIn("Backup job finished", backup["title"]) self.assertIn("Veľkosť: 1.5 GiB", backup["body"]) self.assertIn("Trvanie: 00:00:10", backup["body"]) manager = notification_manager.NotificationManager() manager._config = {"notification_language": "sk"} rows = [(1, "cpu_high", "resources", 0, "pve01: Vysoké využitie CPU", "body")] digest = manager._compose_digest_body(rows) self.assertIn("udalostí INFO zoskupených podľa kategórie", digest) self.assertIn("Zdroje", digest) title, body, caption = manager._build_test_message(False, False, "groq / sk") self.assertEqual(title, "Test ProxMenux") self.assertIn("Vitajte", body) self.assertIn("profilovú fotografiu", caption) def test_batch_app_updates_render_from_every_runtime_catalog(self): data = { "hostname": "HOST-ŽILINA", "updates": [ {"vmid": 100, "app_name": "AdGuard Home", "installed": "1.0", "latest": "1.1"}, {"vmid": 115, "app_name": "Redis", "installed": "7.0", "latest": "8.1"}, ], } for language in self.RUNTIME_LANGUAGES: rendered = notification_templates.render_template( "app_update_available", data, language=language, ) self.assertEqual( rendered["title"], notification_templates.runtime_message( "appUpdates.batchTitle", language, hostname="HOST-ŽILINA", count=2, ), language, ) self.assertIn("CT 100", rendered["body"], language) self.assertIn("• Redis: 7.0 → 8.1", rendered["body"], language) def test_ai_disabled_telegram_receives_deterministic_slovak(self): class RecordingTelegram: def __init__(self): self.payload = None def send(self, title, body, severity, data=None): self.payload = (title, body, severity, data) return {"success": True, "error": ""} channel = RecordingTelegram() manager = notification_manager.NotificationManager() manager._channels = {"telegram": channel} manager._config = { "notification_language": "sk", "ai_enabled": "false", "telegram.rich_format": "false", } with mock.patch.object(manager, "_record_history"): result = manager.send_notification( "vm_start", "INFO", "", "", data={"hostname": "pve01", "vmname": "účtovníctvo", "vmid": "123"}, skip_toggle_check=True, ) self.assertTrue(result["success"]) self.assertIn("spustený", channel.payload[0]) self.assertIn("účtovníctvo", channel.payload[1]) self.assertNotIn("is now running", channel.payload[1]) def test_display_name_replaces_only_the_local_runtime_hostname(self): config = {"hostname": "HomeLAB_2"} with mock.patch.object(notification_manager.socket, "gethostname", return_value="homelab-2"), \ mock.patch.object(notification_manager.socket, "getfqdn", return_value="homelab-2.home.lab"): self.assertEqual( notification_manager.resolve_notification_hostname("homelab-2", config), "HomeLAB_2", ) self.assertEqual( notification_manager.resolve_notification_hostname("remote-pve", config), "remote-pve", ) class RecordingChannel: def __init__(self): self.payload = None def send(self, title, body, severity, data=None): self.payload = (title, body, severity, data) return {"success": True} manager = notification_manager.NotificationManager() channel = RecordingChannel() manager._channels = {"telegram": channel} manager._config = { "notification_language": "sk", "hostname": "HomeLAB_2", "ai_enabled": "false", "telegram.rich_format": "false", } with mock.patch.object(notification_manager.socket, "gethostname", return_value="homelab-2"), \ mock.patch.object(notification_manager.socket, "getfqdn", return_value="homelab-2.home.lab"), \ mock.patch.object(manager, "_record_history"): result = manager.send_notification( "docker_stack_update_available", "INFO", "", "", data={"hostname": "homelab-2", "vmid": "210", "ct_name": "repopulse", "count": "1", "details": "Docker Engine"}, skip_toggle_check=True, ) self.assertTrue(result["success"]) self.assertIn("HomeLAB_2: Na CT 210 sú dostupné aktualizácie Docker", channel.payload[0]) self.assertNotIn("homelab-2:", channel.payload[0]) def test_email_channel_chrome_uses_the_runtime_catalog(self): channel = object.__new__(notification_channels.EmailChannel) channel.subject_prefix = "[ProxMenux]" html = channel._format_html( "[ProxMenux] pve01: Vysoké využitie CPU", "Využitie CPU dosiahlo 95 %.", "WARNING", { "_notification_language": "sk", "_event_type": "cpu_high", "_group": "resources", "hostname": "pve01", "value": "95", "threshold": "90", "cores": "8", }, ) self.assertIn("Systémové zdroje", html) self.assertIn("UPOZORNENIE", html) self.assertIn("Server:", html) self.assertIn("Aktuálna hodnota", html) self.assertNotIn(">Details<", html) self.assertNotIn("System Resources Report", html) vm_html = channel._format_html( "pve01: VM účtovníctvo (123) spustený", "Virtuálny stroj účtovníctvo (ID: 123) je spustený.", "INFO", { "_notification_language": "sk", "_event_type": "vm_start", "_group": "vm_ct", "hostname": "pve01", "vmid": "123", "vmname": "účtovníctvo", }, ) self.assertIn("VM bola spustená", vm_html) def test_notification_language_precedence_and_roundtrip(self): manager = notification_manager.NotificationManager() manager._config = {"notification_language": "sk", "ai_language": "de"} self.assertEqual(manager._notification_language(), "sk") manager._config = {"ai_language": "sk"} self.assertEqual(manager._notification_language(), "sk") manager._config = {} self.assertEqual(manager._notification_language(), "en") 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): saved = manager.save_settings({"notification_language": "sk", "ai_language": "de"}) self.assertTrue(saved["success"]) loaded = notification_manager.NotificationManager() loaded._load_config() self.assertEqual(loaded.get_settings()["config"]["notification_language"], "sk") self.assertEqual(loaded.get_settings()["config"]["ai_language"], "de") def test_ai_language_is_independent_from_runtime_notification_language(self): manager = notification_manager.NotificationManager() manager._config = { "notification_language": "sk", "ai_language": "de", "ai_provider": "groq", } self.assertEqual(manager._notification_language(), "sk") self.assertEqual(manager._build_ai_config()["ai_language"], "de") def test_legacy_runtime_capable_ai_language_is_preserved(self): manager = notification_manager.NotificationManager() manager._config = {"ai_language": "de"} self.assertEqual(manager._notification_language(), "de") self.assertEqual(manager.get_settings()["config"]["notification_language"], "de") manager._config = {"notification_language": "invalid", "ai_language": "sk"} self.assertEqual(manager._notification_language(), "sk") manager._config = {"notification_language": "invalid", "ai_language": "de"} self.assertEqual(manager._notification_language(), "de") 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): result = manager.save_settings({ "notification_language": manager.get_settings()["config"]["notification_language"], "ai_language": "de", }) self.assertTrue(result["success"], result) def test_backup_email_localizes_status_and_important_packages(self): channel = object.__new__(notification_channels.EmailChannel) channel.subject_prefix = "[ProxMenux]" for event_type, localized_status, english_status in ( ("backup_fail", "Zlyhalo", "Failed"), ("backup_complete", "Dokončené", "Completed"), ("backup_start", "Spustené", "Started"), ): backup_html = channel._format_html( "pve01: Záloha", "Stav zálohy VM 100.", "WARNING", { "_notification_language": "sk", "_event_type": event_type, "_group": "backup", "hostname": "pve01", "vmid": "100", "vmname": "alpha", "storage": "pbs-main", }, ) self.assertIn(f">{localized_status}<", backup_html) self.assertNotIn(f">{english_status}<", backup_html) updates_html = channel._format_html( "pve01: Aktualizácie", "Dostupné aktualizácie.", "INFO", { "_notification_language": "sk", "_event_type": "system_updates", "_group": "updates", "hostname": "pve01", "important_list": "pve-manager\nproxmox-kernel", }, ) self.assertIn("Dôležité balíky", updates_html) self.assertNotIn("Important Packages", updates_html) def test_metric_and_system_email_values_are_localized(self): channel = object.__new__(notification_channels.EmailChannel) channel.subject_prefix = "[ProxMenux]" metric_html = channel._format_html( "pve01: Vysoké využitie CPU", "CPU dosiahlo 95 %.", "WARNING", { "_notification_language": "sk", "_event_type": "cpu_high", "_group": "resources", "hostname": "pve01", "value": "95", }, ) self.assertIn("Vysoké využitie CPU", metric_html) self.assertNotIn("Cpu High", metric_html) system_html = channel._format_html( "pve01: Systémový problém", "Zistil sa problém.", "WARNING", { "_notification_language": "sk", "_event_type": "system_problem", "_group": "system", "hostname": "pve01", "reason": "chyba", }, ) self.assertIn("Prehľad: Systém", system_html) self.assertNotIn("Prehľad:

", system_html) def test_every_ui_ai_language_is_accepted_by_backend(self): source = (APPIMAGE_DIR / "components" / "notification-settings.tsx").read_text(encoding="utf-8") block = re.search(r"const AI_LANGUAGES = \[(.*?)\n\]", source, re.DOTALL) self.assertIsNotNone(block) ui_languages = set(re.findall(r'value: "([a-z]+)"', block.group(1))) self.assertIn("sv", ui_languages) self.assertIn("no", ui_languages) self.assertEqual(ui_languages - set(notification_manager.ALLOWED_AI_LANGUAGES), set()) def test_quiet_hours_digest_localizes_system_group(self): class RecordingChannel: def __init__(self): self.payload = None def send(self, title, body, severity, data=None): self.payload = (title, body, severity, data) return {"success": True} manager = notification_manager.NotificationManager() manager._config = {"notification_language": "sk", "hostname": "pve01"} channel = RecordingChannel() with tempfile.TemporaryDirectory() as directory: db_path = Path(directory) / "settings.db" conn = sqlite3.connect(db_path) conn.execute( "CREATE TABLE quiet_pending (" "id INTEGER PRIMARY KEY, channel TEXT, event_type TEXT, event_group TEXT, " "ts REAL, title TEXT, body TEXT)" ) conn.execute( "INSERT INTO quiet_pending " "(channel, event_type, event_group, ts, title, body) VALUES (?, ?, ?, ?, ?, ?)", ("telegram", "ai_model_migrated", "system", 0, "pve01: AI model updated", "body"), ) conn.commit() conn.close() with mock.patch.object(notification_manager, "DB_PATH", db_path), mock.patch.object( manager, "_record_history" ): manager._flush_quiet_for_channel("telegram", channel) self.assertIsNotNone(channel.payload) self.assertIn("Systém: 1", channel.payload[1]) self.assertNotIn("System", channel.payload[1]) self.assertTrue(channel.payload[3]["_quiet_hours_summary"]) def test_visible_templates_use_known_backend_and_frontend_groups(self): visible_groups = { template.get("group", "other") for template in notification_templates.TEMPLATES.values() if not template.get("hidden", False) } backend_groups = set(notification_templates.EVENT_GROUPS) source = (APPIMAGE_DIR / "components" / "notification-settings.tsx").read_text(encoding="utf-8") match = re.search(r"const EVENT_CATEGORIES = \[(.*?)\]\.map", source) self.assertIsNotNone(match) frontend_groups = set(re.findall(r'"([a-z_]+)"', match.group(1))) self.assertEqual(visible_groups - backend_groups, set()) self.assertEqual(visible_groups - frontend_groups, set()) self.assertEqual(frontend_groups, backend_groups) for language in self.RUNTIME_LANGUAGES: catalog = json.loads( (APPIMAGE_DIR / "messages" / language / "common.json").read_text(encoding="utf-8") ) self.assertEqual( set(catalog["settings"]["notifications"]["categories"]), frontend_groups, language, ) def test_build_bundles_runtime_catalogs(self): build = (SCRIPTS_DIR / "build_appimage.sh").read_text(encoding="utf-8") for language in self.RUNTIME_LANGUAGES: self.assertIn(language, build) self.assertIn('$APP_DIR/usr/share/proxmenux/messages', build) def test_missing_event_type_names_exist_in_both_ui_catalogs(self): for language in self.RUNTIME_LANGUAGES: path = APPIMAGE_DIR / "messages" / language / "common.json" event_types = json.loads(path.read_text(encoding="utf-8"))["settings"]["notifications"]["eventTypes"] self.assertIn("lxc_update_applied", event_types) self.assertIn("docker_stack_update_available", event_types) if __name__ == "__main__": unittest.main()