Initial support for mako notification daemon.

This commit is contained in:
2026-04-06 21:09:41 -04:00
parent 91edf64a72
commit a8950c42e2
9 changed files with 1265 additions and 85 deletions

65
tests/conftest.py Normal file
View File

@@ -0,0 +1,65 @@
import importlib.util
import sys
import types
from pathlib import Path
REPO_ROOT = Path(__file__).resolve().parents[1]
SRC_ROOT = REPO_ROOT / "src"
BUILD_PACKAGE_ROOT = REPO_ROOT / "_build" / "src" / "cthulhu"
def _load_generated_module(module_name):
module_path = BUILD_PACKAGE_ROOT / f"{module_name}.py"
full_name = f"cthulhu.{module_name}"
if module_path.exists():
spec = importlib.util.spec_from_file_location(full_name, module_path)
module = importlib.util.module_from_spec(spec)
sys.modules[full_name] = module
spec.loader.exec_module(module)
return module
module = types.ModuleType(full_name)
if module_name == "cthulhu_i18n":
module._ = lambda text: text
module.ngettext = lambda singular, plural, count: singular if count == 1 else plural
module.cgettext = lambda text: text
module.C_ = lambda _context, text: text
module.localedir = str(REPO_ROOT / "po")
module.setModuleLocale = lambda *_args, **_kwargs: None
module.setLocaleForMessages = lambda *_args, **_kwargs: None
module.setLocaleForNames = lambda *_args, **_kwargs: None
module.setLocaleForGUI = lambda *_args, **_kwargs: None
return module
if module_name == "cthulhu_platform":
from cthulhu import cthulhuVersion
module.version = (
f"Cthulhu screen reader version "
f"{cthulhuVersion.version}-{cthulhuVersion.codeName}"
)
module.revision = ""
module.prefix = str(REPO_ROOT)
module.package = "cthulhu"
module.datadir = str(REPO_ROOT)
module.tablesdir = "/usr/share/liblouis/tables"
return module
raise ImportError(f"Unsupported generated module: {module_name}")
if str(SRC_ROOT) not in sys.path:
sys.path.insert(0, str(SRC_ROOT))
import cthulhu # noqa: E402
for generated_module in ("cthulhu_i18n", "cthulhu_platform"):
if f"cthulhu.{generated_module}" in sys.modules:
continue
loaded_module = _load_generated_module(generated_module)
sys.modules[f"cthulhu.{generated_module}"] = loaded_module
setattr(cthulhu, generated_module, loaded_module)

View File

@@ -0,0 +1,160 @@
import sys
import unittest
from pathlib import Path
from unittest import mock
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
from cthulhu import notification_presenter
from cthulhu.mako_notification_monitor import MakoNotificationMonitor
class _FakeScript:
def __init__(self):
self.speechGenerator = mock.Mock()
self.speechGenerator.voice.return_value = object()
self.speakMessage = mock.Mock()
self.displayBrailleMessage = mock.Mock()
class _FakeScriptManager:
def __init__(self, script):
self._script = script
def get_default_script(self):
return self._script
class _FakeApp:
def __init__(self, script):
self._script_manager = _FakeScriptManager(script)
def getScriptManager(self):
return self._script_manager
class MakoNotificationMonitorTests(unittest.TestCase):
def setUp(self):
self.presenter = notification_presenter.NotificationPresenter()
self.script = _FakeScript()
self.app = _FakeApp(self.script)
presenter_patcher = mock.patch(
"cthulhu.mako_notification_monitor.notification_presenter.getPresenter",
return_value=self.presenter,
)
self.addCleanup(presenter_patcher.stop)
presenter_patcher.start()
self.monitor = MakoNotificationMonitor(self.app)
self.monitor._generation = 1
def test_initial_seed_does_not_announce_existing_notifications(self):
self.monitor._sync_notifications(
[
{
"id": 10,
"app-name": "discord",
"summary": "hello",
"body": "world",
"actions": {"default": "View"},
}
],
announce_new=False,
)
self.assertEqual(self.monitor._known_ids, {10})
self.assertEqual(self.presenter._notifications, [])
self.script.speakMessage.assert_not_called()
self.script.displayBrailleMessage.assert_not_called()
def test_new_notification_is_spoken_and_saved(self):
self.monitor._sync_notifications(
[
{
"id": 10,
"app-name": "discord",
"summary": "old",
"body": "",
"actions": {},
}
],
announce_new=False,
)
self.monitor._sync_notifications(
[
{
"id": 10,
"app-name": "discord",
"summary": "old",
"body": "",
"actions": {},
},
{
"id": 11,
"app-name": "discord",
"summary": "new summary",
"body": "new body",
"actions": {"default": "View"},
},
],
announce_new=True,
)
self.assertEqual(len(self.presenter._notifications), 1)
entry = self.presenter._notifications[0]
self.assertEqual(entry.notification_id, 11)
self.assertTrue(entry.live)
self.assertEqual(entry.source, "mako")
self.assertEqual(entry.actions, {"default": "View"})
self.assertEqual(entry.message, "Notification new summary new body")
self.script.speakMessage.assert_called_once()
self.script.displayBrailleMessage.assert_called_once()
def test_removed_notification_is_marked_not_live(self):
self.monitor._sync_notifications(
[
{
"id": 11,
"app-name": "discord",
"summary": "new summary",
"body": "new body",
"actions": {"default": "View"},
},
],
announce_new=True,
)
self.assertTrue(self.presenter._notifications[0].live)
self.monitor._sync_notifications([], announce_new=False)
self.assertFalse(self.presenter._notifications[0].live)
self.assertEqual(self.monitor._known_ids, set())
def test_parse_notification_strips_markup_and_normalizes_actions(self):
parsed = self.monitor._parse_notification(
{
"id": 22,
"app-name": "Crash Reporting System",
"summary": "<b>Service Crash</b>",
"body": "<html><tt>/usr/bin/python3.14</tt> crashed</html>",
"desktop-entry": "crash-handler",
"urgency": 2,
"actions": {"1": "<b>Details</b>"},
}
)
self.assertEqual(parsed["notification_id"], 22)
self.assertEqual(parsed["summary"], "Service Crash")
self.assertEqual(parsed["body"], "/usr/bin/python3.14 crashed")
self.assertEqual(parsed["actions"], {"1": "Details"})
self.assertEqual(
parsed["message"],
"Notification Service Crash /usr/bin/python3.14 crashed",
)
if __name__ == "__main__":
unittest.main()

View File

@@ -0,0 +1,124 @@
import sys
import unittest
from pathlib import Path
from unittest import mock
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
from cthulhu import messages
from cthulhu.notification_presenter import NotificationPresenter
class _FakeMonitor:
def __init__(self):
self.dismissed = []
self.invoked = []
self.current_generation = 2
def refresh(self, announce_new=False):
return True
def is_current_entry(self, entry):
return entry.source == "mako" and entry.source_generation == self.current_generation
def dismiss_notification(self, notification_id):
self.dismissed.append(notification_id)
return True
def invoke_action(self, notification_id, action_key):
self.invoked.append((notification_id, action_key))
return True
class NotificationPresenterMakoTests(unittest.TestCase):
def setUp(self):
self.presenter = NotificationPresenter()
self.monitor = _FakeMonitor()
self.presenter.set_mako_monitor(self.monitor)
def test_sync_updates_current_generation_only(self):
old_entry = self.presenter.save_notification(
"Notification old",
source="mako",
source_generation=1,
notification_id=5,
live=False,
actions={"default": "Old"},
)
current_entry = self.presenter.save_notification(
"Notification current",
source="mako",
source_generation=2,
notification_id=6,
live=True,
actions={"default": "View"},
)
self.presenter.sync_live_notifications(
"mako",
{
5: {
"message": "Notification replaced",
"actions": {"default": "New"},
"app_name": "discord",
"summary": "replaced",
"body": "",
"urgency": 1,
"desktop_entry": "discord",
}
},
source_generation=2,
)
self.assertEqual(old_entry.message, "Notification old")
self.assertEqual(old_entry.actions, {"default": "Old"})
self.assertFalse(current_entry.live)
def test_can_control_entry_and_get_actions_require_current_monitor_generation(self):
entry = self.presenter.save_notification(
"Notification current",
source="mako",
source_generation=2,
notification_id=6,
live=True,
actions={"default": "View"},
)
self.assertTrue(self.presenter.can_control_entry(entry))
self.assertEqual(self.presenter.get_actions_for_entry(entry), {"default": "View"})
stale_entry = self.presenter.save_notification(
"Notification stale",
source="mako",
source_generation=1,
notification_id=7,
live=True,
actions={"default": "Open"},
)
self.assertFalse(self.presenter.can_control_entry(stale_entry))
self.assertEqual(self.presenter.get_actions_for_entry(stale_entry), {})
def test_dismiss_and_invoke_action_route_through_monitor(self):
entry = self.presenter.save_notification(
"Notification current",
source="mako",
source_generation=2,
notification_id=6,
live=True,
actions={"default": "View"},
)
script = mock.Mock()
self.assertTrue(self.presenter.dismiss_entry(script, entry))
self.assertEqual(self.monitor.dismissed, [6])
script.presentMessage.assert_called_with(messages.NOTIFICATION_DISMISSED)
script.reset_mock()
self.assertTrue(self.presenter.invoke_action_for_entry(script, entry, "default"))
self.assertEqual(self.monitor.invoked, [(6, "default")])
script.presentMessage.assert_called_with(messages.NOTIFICATION_ACTION_INVOKED)
if __name__ == "__main__":
unittest.main()