91 lines
3.3 KiB
Python
91 lines
3.3 KiB
Python
"""dbus-monitor parsing + the forward/rate-limit gate. No session bus
|
|
needed — the parser is fed canned dbus-monitor output."""
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
|
|
|
from bolt_pet.notifications import Notification, NotificationGate, iter_notifications
|
|
|
|
SAMPLE = '''signal time=1710000000.1 sender=org.freedesktop.DBus -> destination=:1.7 serial=2 path=/org/freedesktop/DBus; interface=org.freedesktop.DBus; member=NameAcquired
|
|
string ":1.7"
|
|
method call time=1710000001.2 sender=:1.72 -> destination=org.freedesktop.Notifications serial=88 path=/org/freedesktop/Notifications; interface=org.freedesktop.Notifications; member=Notify
|
|
string "Firefox"
|
|
uint32 0
|
|
string ""
|
|
string "Build finished"
|
|
string "All 42 tests passed"
|
|
array [
|
|
]
|
|
int32 -1
|
|
method call time=1710000002.3 sender=:1.80 -> destination=org.freedesktop.Notifications serial=91 path=/org/freedesktop/Notifications; interface=org.freedesktop.Notifications; member=Notify
|
|
string "Calendar"
|
|
uint32 0
|
|
string "calendar-icon"
|
|
string "Standup in 5 minutes"
|
|
string ""
|
|
array [
|
|
]
|
|
'''
|
|
|
|
|
|
def _parse(text=SAMPLE):
|
|
return list(iter_notifications(text.splitlines()))
|
|
|
|
|
|
def test_parses_each_notify_call():
|
|
parsed = _parse()
|
|
assert parsed == [
|
|
Notification(app="Firefox", summary="Build finished", body="All 42 tests passed"),
|
|
Notification(app="Calendar", summary="Standup in 5 minutes", body=""),
|
|
]
|
|
|
|
|
|
def test_unrelated_dbus_traffic_is_ignored():
|
|
noise = SAMPLE.split("method call")[0]
|
|
assert _parse(noise) == []
|
|
|
|
|
|
def test_trailing_notification_without_a_following_block_is_still_emitted():
|
|
assert _parse()[-1].summary == "Standup in 5 minutes"
|
|
|
|
|
|
def test_as_text_is_what_gets_sent_to_the_server():
|
|
assert _parse()[0].as_text() == "Firefox: Build finished — All 42 tests passed"
|
|
assert _parse()[1].as_text() == "Calendar: Standup in 5 minutes"
|
|
|
|
|
|
# ── gate ────────────────────────────────────────────────────────────────────
|
|
|
|
def _notification(summary="Build finished", app="Firefox"):
|
|
return Notification(app=app, summary=summary, body="")
|
|
|
|
|
|
def test_empty_filter_forwards_everything():
|
|
gate = NotificationGate("", min_interval=0)
|
|
assert gate.should_forward(_notification(), now=0) is True
|
|
|
|
|
|
def test_filter_regex_selects_what_is_worth_a_round_trip():
|
|
gate = NotificationGate(r"build|deploy", min_interval=0)
|
|
assert gate.should_forward(_notification("Build finished"), now=0) is True
|
|
assert gate.should_forward(_notification("New message from Dave"), now=1) is False
|
|
|
|
|
|
def test_rate_limit_drops_a_burst():
|
|
gate = NotificationGate("", min_interval=60)
|
|
assert gate.should_forward(_notification(), now=100) is True
|
|
assert gate.should_forward(_notification(), now=120) is False
|
|
assert gate.should_forward(_notification(), now=161) is True
|
|
|
|
|
|
def test_a_broken_regex_does_not_silence_the_bridge():
|
|
gate = NotificationGate("(unclosed", min_interval=0)
|
|
assert gate.should_forward(_notification(), now=0) is True
|
|
|
|
|
|
def test_empty_notifications_are_dropped():
|
|
gate = NotificationGate("", min_interval=0)
|
|
assert gate.should_forward(Notification(app="", summary="", body=""), now=0) is False
|