1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
|
# This program is free software; you can redistribute it and/or modify it under
# the terms of the GNU Lesser General Public License as published by the Free
# Software Foundation; either version 3 of the License, or (at your option) any
# later version. See http://www.gnu.org/copyleft/lgpl.html for the full text
# of the license.
__author__ = "Martin Pitt"
__copyright__ = """
(c) 2012 Canonical Ltd.
(c) 2017 - 2022 Martin Pitt <martin@piware.de>
"""
import fcntl
import os
import subprocess
import sys
import unittest
import dbus
import dbusmock
try:
notify_send_version = subprocess.check_output(["notify-send", "--version"], text=True)
notify_send_version = notify_send_version.split()[-1]
except (OSError, subprocess.CalledProcessError):
notify_send_version = ""
@unittest.skipUnless(notify_send_version, "notify-send not installed")
class TestNotificationDaemon(dbusmock.DBusTestCase):
"""Test mocking notification-daemon"""
@classmethod
def setUpClass(cls):
cls.start_session_bus()
cls.dbus_con = cls.get_dbus(False)
def setUp(self):
(self.p_mock, self.obj_daemon) = self.spawn_server_template("notification_daemon", {}, stdout=subprocess.PIPE)
# set log to nonblocking
flags = fcntl.fcntl(self.p_mock.stdout, fcntl.F_GETFL)
fcntl.fcntl(self.p_mock.stdout, fcntl.F_SETFL, flags | os.O_NONBLOCK)
def tearDown(self):
self.p_mock.stdout.close()
self.p_mock.terminate()
self.p_mock.wait()
def test_no_options(self):
"""notify-send with no options"""
subprocess.check_call(["notify-send", "title", "my text"])
log = self.p_mock.stdout.read()
self.assertRegex(log, b'[0-9.]+ Notify "notify-send" 0 "" "title" "my text" \\[\\]')
def test_options(self):
"""notify-send with some options"""
subprocess.check_call(["notify-send", "-t", "27", "-a", "fooApp", "-i", "warning_icon", "title", "my text"])
log = self.p_mock.stdout.read()
# HACK: Why is the timeout missing on s390x?
if os.uname().machine != "s390x":
self.assertRegex(log, rb"[0-9.]+ Notify.* 27\n")
# libnotify 0.8.4 changes warning_icon from positional to optional argument, so check that separately
self.assertRegex(log, rb'[0-9.]+ Notify "fooApp" 0 "(warning_icon)?" "title" "my text" \[\] {.*"urgency": 1')
self.assertRegex(log, rb"[0-9.]+ Notify .*warning_icon")
def test_id(self):
"""ID handling"""
notify_proxy = dbus.Interface(
self.dbus_con.get_object("org.freedesktop.Notifications", "/org/freedesktop/Notifications"),
"org.freedesktop.Notifications",
)
# with input ID 0 it should generate new IDs
id_ = notify_proxy.Notify("test", 0, "", "summary", "body", [], {}, -1)
self.assertEqual(id_, 1)
id_ = notify_proxy.Notify("test", 0, "", "summary", "body", [], {}, -1)
self.assertEqual(id_, 2)
# an existing ID should just be bounced back
id_ = notify_proxy.Notify("test", 4, "", "summary", "body", [], {}, -1)
self.assertEqual(id_, 4)
id_ = notify_proxy.Notify("test", 1, "", "summary", "body", [], {}, -1)
self.assertEqual(id_, 1)
# the previous doesn't forget the counter
id_ = notify_proxy.Notify("test", 0, "", "summary", "body", [], {}, -1)
self.assertEqual(id_, 3)
def test_close(self):
"""CloseNotification() and NotificationClosed() signal"""
notify_proxy = dbus.Interface(
self.dbus_con.get_object("org.freedesktop.Notifications", "/org/freedesktop/Notifications"),
"org.freedesktop.Notifications",
)
id_ = notify_proxy.Notify("test", 0, "", "summary", "body", [], {}, -1)
self.assertEqual(id_, 1)
# known notification, should send a signal
notify_proxy.CloseNotification(id_)
log = self.p_mock.stdout.read()
self.assertRegex(log, b"[0-9.]+ emit .*NotificationClosed 1 1\n")
# unknown notification, don't send a signal
notify_proxy.CloseNotification(id_ + 1)
log = self.p_mock.stdout.read()
self.assertNotIn(b"NotificationClosed", log)
if __name__ == "__main__":
# avoid writing to stderr
unittest.main(testRunner=unittest.TextTestRunner(stream=sys.stdout))
|
