testing: more tests for notifications
Also cover temporary network error (no notification) and successful sync (start and end notifications).
This commit is contained in:
Patrick Ohly
parent
430de7a7ed
commit
5427b186df
|
|
@ -52,6 +52,12 @@ monitor = ["dbus-monitor"]
|
|||
xdg_root = "temp-test-dbus"
|
||||
configName = "dbus_unittest"
|
||||
|
||||
def GrepNotifications(dbuslog):
|
||||
'''finds all Notify calls and returns their parameters as list of line lists'''
|
||||
return re.findall(r'^method call .* dest=org.freedesktop.Notifications .*interface=org.freedesktop.Notifications; member=Notify\n((?:^ .*\n)*)',
|
||||
dbuslog,
|
||||
re.MULTILINE)
|
||||
|
||||
# See notification-daemon.py for a stand-alone version.
|
||||
#
|
||||
# Embedded here to avoid issues with setting up the environment
|
||||
|
|
@ -1959,11 +1965,11 @@ class TestSessionAPIsDummy(unittest.TestCase, DBusUtil):
|
|||
self.failUnlessEqual(self.lastState, "done")
|
||||
|
||||
@timeout(60)
|
||||
def testAutoSyncFailure(self):
|
||||
"""TestSessionAPIsDummy.testAutoSyncFailure - test that auto-sync is triggered, fails here"""
|
||||
def testAutoSyncNetworkFailure(self):
|
||||
"""TestSessionAPIsDummy.testAutoSyncNetworkFailure - test that auto-sync is triggered, fails due to (temporary?!) network error here"""
|
||||
self.setupConfig()
|
||||
# enable auto-sync
|
||||
config = self.config
|
||||
config = copy.deepcopy(self.config)
|
||||
# Note that writing this config will modify the host's keyring!
|
||||
# Use a syncURL that is unlikely to conflict with the host
|
||||
# or any other D-Bus test.
|
||||
|
|
@ -2027,12 +2033,23 @@ class TestSessionAPIsDummy(unittest.TestCase, DBusUtil):
|
|||
self.failUnless(delta < 13)
|
||||
self.failUnless(delta > 7)
|
||||
|
||||
# check that org.freedesktop.Notifications.Notify was not called
|
||||
# (network errors are considered temporary, can't tell in this case
|
||||
# that the name lookup error is permanent)
|
||||
def checkDBusLog(self, content):
|
||||
notifications = GrepNotifications(content)
|
||||
self.failUnlessEqual(notifications,
|
||||
[])
|
||||
|
||||
# done as part of post-processing in runTest()
|
||||
self.runTestDBusCheck = checkDBusLog
|
||||
|
||||
@timeout(60)
|
||||
def testAutoSyncLocal(self):
|
||||
"""TestSessionAPIsDummy.testAutoSyncLocal - test that auto-sync is triggered for local sync"""
|
||||
def testAutoSyncLocalConfigError(self):
|
||||
"""TestSessionAPIsDummy.testAutoSyncLocalConfigError - test that auto-sync is triggered for local sync, fails due to permanent config error here"""
|
||||
self.setupConfig()
|
||||
# enable auto-sync
|
||||
config = self.config
|
||||
config = copy.deepcopy(self.config)
|
||||
config[""]["syncURL"] = "local://@foobar" # will fail
|
||||
config[""]["autoSync"] = "1"
|
||||
config[""]["autoSyncDelay"] = "0"
|
||||
|
|
@ -2076,10 +2093,9 @@ class TestSessionAPIsDummy(unittest.TestCase, DBusUtil):
|
|||
self.failUnlessEqual(flags, [])
|
||||
|
||||
# check that org.freedesktop.Notifications.Notify was called
|
||||
# once to report the failed attempt to start the sync
|
||||
def checkDBusLog(self, content):
|
||||
notifications = re.findall(r'^method call .* dest=org.freedesktop.Notifications .*interface=org.freedesktop.Notifications; member=Notify\n((?:^ .*\n)*)',
|
||||
content,
|
||||
re.MULTILINE)
|
||||
notifications = GrepNotifications(content)
|
||||
self.failUnlessEqual(notifications,
|
||||
[' string "SyncEvolution"\n'
|
||||
' uint32 0\n'
|
||||
|
|
@ -2099,6 +2115,106 @@ class TestSessionAPIsDummy(unittest.TestCase, DBusUtil):
|
|||
# done as part of post-processing in runTest()
|
||||
self.runTestDBusCheck = checkDBusLog
|
||||
|
||||
@timeout(60)
|
||||
def testAutoSyncLocalSuccess(self):
|
||||
"""TestSessionAPIsDummy.testAutoSyncLocalSuccess - test that auto-sync is done successfully for local sync between file backends"""
|
||||
# create @foobar config
|
||||
self.session.Detach()
|
||||
self.setUpSession("target-config@foobar")
|
||||
config = copy.deepcopy(self.config)
|
||||
config[""]["remoteDeviceId"] = "foo"
|
||||
config[""]["deviceId"] = "bar"
|
||||
for i in ("addressbook", "calendar", "todo", "memo"):
|
||||
source = config["source/" + i]
|
||||
source["database"] = source["database"] + ".server"
|
||||
self.session.SetConfig(False, False, config, utf8_strings=True)
|
||||
self.session.Detach()
|
||||
|
||||
# create dummy-test@default auto-sync config
|
||||
self.setUpSession("dummy-test")
|
||||
config = copy.deepcopy(self.config)
|
||||
config[""]["syncURL"] = "local://@foobar"
|
||||
config[""]["PeerIsClient"] = "1"
|
||||
config[""]["autoSync"] = "1"
|
||||
config[""]["autoSyncDelay"] = "0"
|
||||
config[""]["autoSyncInterval"] = "10s"
|
||||
config["source/addressbook"]["uri"] = "addressbook"
|
||||
self.session.SetConfig(False, False, config, utf8_strings=True)
|
||||
|
||||
def session_ready(object, ready):
|
||||
if self.running and object != self.sessionpath and \
|
||||
(self.auto_sync_session_path == None and ready or \
|
||||
self.auto_sync_session_path == object):
|
||||
self.auto_sync_session_path = object
|
||||
DBusUtil.quit_events.append("session " + object + (ready and " ready" or " done"))
|
||||
loop.quit()
|
||||
|
||||
signal = bus.add_signal_receiver(session_ready,
|
||||
'SessionChanged',
|
||||
'org.syncevolution.Server',
|
||||
self.server.bus_name,
|
||||
None,
|
||||
byte_arrays=True,
|
||||
utf8_strings=True)
|
||||
|
||||
# shut down current session, will allow auto-sync
|
||||
self.session.Detach()
|
||||
|
||||
# wait for start and end of auto-sync session
|
||||
loop.run()
|
||||
loop.run()
|
||||
self.failUnlessEqual(DBusUtil.quit_events, ["session " + self.auto_sync_session_path + " ready",
|
||||
"session " + self.auto_sync_session_path + " done"])
|
||||
session = dbus.Interface(bus.get_object(self.server.bus_name,
|
||||
self.auto_sync_session_path),
|
||||
'org.syncevolution.Session')
|
||||
reports = session.GetReports(0, 100, utf8_strings=True)
|
||||
self.failUnlessEqual(len(reports), 1)
|
||||
self.failUnlessEqual(reports[0]["status"], "200")
|
||||
name = session.GetConfigName()
|
||||
self.failUnlessEqual(name, "dummy-test")
|
||||
flags = session.GetFlags()
|
||||
self.failUnlessEqual(flags, [])
|
||||
|
||||
# check that org.freedesktop.Notifications.Notify was called
|
||||
# when starting and completing the sync
|
||||
def checkDBusLog(self, content):
|
||||
notifications = GrepNotifications(content)
|
||||
self.failUnlessEqual(notifications,
|
||||
[' string "SyncEvolution"\n'
|
||||
' uint32 0\n'
|
||||
' string ""\n'
|
||||
' string "dummy-test is syncing"\n'
|
||||
' string "We have just started to sync your computer with the dummy-test sync service."\n'
|
||||
' array [\n'
|
||||
' string "view"\n'
|
||||
' string "View"\n'
|
||||
' string "default"\n'
|
||||
' string "Dismiss"\n'
|
||||
' ]\n'
|
||||
' array [\n'
|
||||
' ]\n'
|
||||
' int32 -1\n',
|
||||
|
||||
' string "SyncEvolution"\n'
|
||||
' uint32 0\n'
|
||||
' string ""\n'
|
||||
' string "dummy-test sync complete"\n'
|
||||
' string "We have just finished syncing your computer with the dummy-test sync service."\n'
|
||||
' array [\n'
|
||||
' string "view"\n'
|
||||
' string "View"\n'
|
||||
' string "default"\n'
|
||||
' string "Dismiss"\n'
|
||||
' ]\n'
|
||||
' array [\n'
|
||||
' ]\n'
|
||||
' int32 -1\n'])
|
||||
|
||||
# done as part of post-processing in runTest()
|
||||
self.runTestDBusCheck = checkDBusLog
|
||||
|
||||
|
||||
class TestSessionAPIsReal(unittest.TestCase, DBusUtil):
|
||||
""" This class is used to test those unit tests of session APIs, depending on doing sync.
|
||||
Thus we need a real server configuration to confirm sync could be run successfully.
|
||||
|
|
|
|||
Loading…
Reference in New Issue