1
0
mirror of https://github.com/arsenetar/dupeguru.git synced 2024-11-09 01:29:01 +00:00
dupeguru/hscommon/tests/notify_test.py

156 lines
4.3 KiB
Python
Raw Normal View History

2019-09-10 00:54:28 +00:00
# Copyright 2015 Hardcoded Software (http://www.hardcoded.net)
# This software is licensed under the "GPLv3" License as described in the "LICENSE" file,
# which should be included with this package. The terms are also available at
2019-09-10 00:54:28 +00:00
# http://www.gnu.org/licenses/gpl-3.0.html
2022-05-09 06:40:08 +00:00
from hscommon.testutil import eq_
from hscommon.notify import Broadcaster, Listener, Repeater
2019-09-10 00:54:28 +00:00
2019-09-10 00:54:28 +00:00
class HelloListener(Listener):
def __init__(self, broadcaster):
Listener.__init__(self, broadcaster)
self.hello_count = 0
def hello(self):
self.hello_count += 1
2019-09-10 00:54:28 +00:00
class HelloRepeater(Repeater):
def __init__(self, broadcaster):
Repeater.__init__(self, broadcaster)
self.hello_count = 0
def hello(self):
self.hello_count += 1
2019-09-10 00:54:28 +00:00
def create_pair():
b = Broadcaster()
listener = HelloListener(b)
return b, listener
2019-09-10 00:54:28 +00:00
2019-09-10 00:54:28 +00:00
def test_disconnect_during_notification():
# When a listener disconnects another listener the other listener will not receive a
2019-09-10 00:54:28 +00:00
# notification.
# This whole complication scheme below is because the order of the notification is not
# guaranteed. We could disconnect everything from self.broadcaster.listeners, but this
# member is supposed to be private. Hence, the '.other' scheme
class Disconnecter(Listener):
def __init__(self, broadcaster):
Listener.__init__(self, broadcaster)
self.hello_count = 0
2019-09-10 00:54:28 +00:00
def hello(self):
self.hello_count += 1
self.other.disconnect()
2019-09-10 00:54:28 +00:00
broadcaster = Broadcaster()
first = Disconnecter(broadcaster)
second = Disconnecter(broadcaster)
first.other, second.other = second, first
first.connect()
second.connect()
broadcaster.notify("hello")
2019-09-10 00:54:28 +00:00
# only one of them was notified
eq_(first.hello_count + second.hello_count, 1)
2019-09-10 00:54:28 +00:00
def test_disconnect():
# After a disconnect, the listener doesn't hear anything.
b, listener = create_pair()
listener.connect()
listener.disconnect()
b.notify("hello")
eq_(listener.hello_count, 0)
2019-09-10 00:54:28 +00:00
2019-09-10 00:54:28 +00:00
def test_disconnect_when_not_connected():
# When disconnecting an already disconnected listener, nothing happens.
b, listener = create_pair()
listener.disconnect()
2019-09-10 00:54:28 +00:00
2019-09-10 00:54:28 +00:00
def test_not_connected_on_init():
# A listener is not initialized connected.
b, listener = create_pair()
b.notify("hello")
eq_(listener.hello_count, 0)
2019-09-10 00:54:28 +00:00
2019-09-10 00:54:28 +00:00
def test_notify():
# The listener listens to the broadcaster.
b, listener = create_pair()
listener.connect()
b.notify("hello")
eq_(listener.hello_count, 1)
2019-09-10 00:54:28 +00:00
2019-09-10 00:54:28 +00:00
def test_reconnect():
# It's possible to reconnect a listener after disconnection.
b, listener = create_pair()
listener.connect()
listener.disconnect()
listener.connect()
b.notify("hello")
eq_(listener.hello_count, 1)
2019-09-10 00:54:28 +00:00
2019-09-10 00:54:28 +00:00
def test_repeater():
b = Broadcaster()
r = HelloRepeater(b)
listener = HelloListener(r)
2019-09-10 00:54:28 +00:00
r.connect()
listener.connect()
b.notify("hello")
2019-09-10 00:54:28 +00:00
eq_(r.hello_count, 1)
eq_(listener.hello_count, 1)
2019-09-10 00:54:28 +00:00
2019-09-10 00:54:28 +00:00
def test_repeater_with_repeated_notifications():
# If REPEATED_NOTIFICATIONS is not empty, only notifs in this set are repeated (but they're
# still dispatched locally).
class MyRepeater(HelloRepeater):
2022-04-28 01:53:12 +00:00
REPEATED_NOTIFICATIONS = {"hello"}
2019-09-10 00:54:28 +00:00
def __init__(self, broadcaster):
HelloRepeater.__init__(self, broadcaster)
self.foo_count = 0
2019-09-10 00:54:28 +00:00
def foo(self):
self.foo_count += 1
2019-09-10 00:54:28 +00:00
b = Broadcaster()
r = MyRepeater(b)
listener = HelloListener(r)
2019-09-10 00:54:28 +00:00
r.connect()
listener.connect()
b.notify("hello")
b.notify("foo") # if the repeater repeated this notif, we'd get a crash on HelloListener
2019-09-10 00:54:28 +00:00
eq_(r.hello_count, 1)
eq_(listener.hello_count, 1)
2019-09-10 00:54:28 +00:00
eq_(r.foo_count, 1)
2019-09-10 00:54:28 +00:00
def test_repeater_doesnt_try_to_dispatch_to_self_if_it_cant():
# if a repeater doesn't handle a particular message, it doesn't crash and simply repeats it.
b = Broadcaster()
r = Repeater(b) # doesnt handle hello
listener = HelloListener(r)
2019-09-10 00:54:28 +00:00
r.connect()
listener.connect()
b.notify("hello") # no crash
eq_(listener.hello_count, 1)
2019-09-10 00:54:28 +00:00
2019-09-10 00:54:28 +00:00
def test_bind_messages():
b, listener = create_pair()
listener.bind_messages({"foo", "bar"}, listener.hello)
listener.connect()
b.notify("foo")
b.notify("bar")
b.notify("hello") # Normal dispatching still work
eq_(listener.hello_count, 3)