diff --git a/fimfarchive/signals.py b/fimfarchive/signals.py new file mode 100644 index 0000000..da4b512 --- /dev/null +++ b/fimfarchive/signals.py @@ -0,0 +1,171 @@ +""" +Signals for Fimfarchive. +""" + + +# +# Fimfarchive, preserves stories from Fimfiction. +# Copyright (C) 2015 Joakim Soderlund +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . +# + + +import blinker + + +__all__ = ( + 'Signal', + 'SignalBinder', + 'SignalSender', + 'SignalReceiver', +) + + +class Signal(blinker.Signal): + """ + Blinker signal with positional arguments. + """ + + def __init__(self, *spec): + """ + Constructor. + + Args: + spec: Names of the signal arguments. + """ + if 'sender' in spec: + raise ValueError("Reserved argument name: 'sender'") + + self.spec = ('sender', *spec) + super().__init__(doc=repr(self)) + + def __call__(self, *args, **kwargs): + """ + Raises an error regarding unbound signals. + """ + raise ValueError( + "Unbound signal. Forgot {}'s initializer?" + .format(type(SignalSender).__name__) + ) + + def __repr__(self): + return "".format(', '.join(self.spec)) + + def send(self, *args, **kwargs): + """ + Emits this signal on behalf of its sender. + + Args: + sender: Object sending the signal. + *args: Values to pass to the receiver. + + Returns: + A list of 2-tuples. Each tuple contains the + signal's receiver and its returned values. + """ + + if len(self.spec) < len(args): + raise ValueError( + "Expected at most {} arguments, got {}." + .format(len(self.spec), len(args)) + ) + + data = {self.spec[i]: v for i, v in enumerate(args)} + duplicates = set(data.keys()).intersection(kwargs.keys()) + + if duplicates: + raise ValueError( + "Got duplicate values for: '{}'" + .format("', '".join(duplicates)) + ) + + data.update(kwargs) + sender = data.pop('sender', None) + + return super().send(sender, **data) + + +class SignalBinder: + """ + Bound transparent proxy for signals. + """ + + def __init__(self, signal, sender): + """ + Constructor. + + Args: + signal: Object to bind for. + sender: Object to bind to. + """ + self.signal = signal + self.sender = sender + + def __call__(self, *args, **kwargs): + return self.send(self.sender, *args, **kwargs) + + def __getattr__(self, attr): + return getattr(self.signal, attr) + + def __repr__(self): + return "".format(self.signal, self.sender) + + +class SignalSender: + """ + Automatically binds signals on init. + """ + + def __init__(self): + """ + Constructor. + """ + sources = { + k: v for k, v in vars(type(self)).items() + if k.startswith('on_') and isinstance(v, Signal) + } + + for k, v in sources.items(): + setattr(self, k, SignalBinder(v, self)) + + +class SignalReceiver: + """ + Automatically connects signals on init. + """ + + def __init__(self, sender): + """ + Constructor. + + Args: + sender: Object to connect to. + """ + sources = { + k for k, v in vars(type(sender)).items() + if k.startswith('on_') and isinstance(v, Signal) + } + + targets = { + k for k, v in vars(type(self)).items() + if k.startswith('on_') and callable(v) + } + + connect = sources.intersection(targets) + + for key in connect: + method = getattr(self, key) + signal = getattr(sender, key) + signal.connect(method, sender=sender) diff --git a/tests/test_signals.py b/tests/test_signals.py new file mode 100644 index 0000000..10c50ff --- /dev/null +++ b/tests/test_signals.py @@ -0,0 +1,188 @@ +""" +Signal tests. +""" + + +# +# Fimfarchive, preserves stories from Fimfiction. +# Copyright (C) 2015 Joakim Soderlund +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . +# + + +from collections import OrderedDict +from unittest.mock import patch, Mock + +import blinker +import pytest + +from fimfarchive.signals import ( + Signal, SignalBinder, SignalSender, SignalReceiver +) + + +@pytest.fixture +def params(): + """ + Returns an ordered dict of parameters. + """ + data = ( + ('a', 1), + ('b', 2), + ('c', 3), + ) + + return OrderedDict(data) + + +@pytest.fixture +def signal(params): + """ + Returns an unbound signal instance. + """ + return Signal(*params.keys()) + + +@pytest.fixture +def sender(signal): + """ + Returns a signal sender instance. + """ + class Sender(SignalSender): + on_signal = signal + + return Sender() + + +@pytest.fixture +def receiver(sender): + """ + Returns a signal receiver instance. + """ + class Receiver(SignalReceiver): + on_signal = Mock('on_signal') + + return Receiver(sender) + + +@pytest.fixture +def binder(sender): + """ + Returns a bound signal instance. + """ + return sender.on_signal + + +class TestSignal: + """ + Signal tests. + """ + + def test_reserved_value_names(self): + """ + Tests `ValueError` is raised for reserved value names. + """ + with pytest.raises(ValueError): + Signal('sender') + + def test_send_unbound_signal(self, params, signal): + """ + Tests `ValueError` is raised when calling unbound signals. + """ + with pytest.raises(ValueError): + signal(*params.values()) + + def test_parameter_mapping(self, params, signal, sender): + """ + Tests positional parameters maps to named parameters. + """ + with patch.object(blinker.Signal, 'send') as m: + signal.send(sender, *params.values()) + m.assert_called_once_with(sender, **params) + + def test_parameter_overflow(self, params, signal, sender): + """ + Tests `ValueError` is raised on too many parameters. + """ + with pytest.raises(ValueError): + signal.send(sender, *params.values(), 'alpaca') + + def test_duplicate_parameter(self, params, signal, sender): + """ + Tests `ValueError` is raised on duplicate parameters. + """ + duplicate = dict(tuple(params.items())[:1]) + + with pytest.raises(ValueError): + signal.send(sender, *params.values(), **duplicate) + + +class TestSignalBinder: + """ + SignalBinder tests. + """ + + def test_send(self, params, sender, binder): + """ + Tests sender is passed to signal. + """ + with patch.object(Signal, 'send') as m: + binder(*params.values()) + m.assert_called_once_with(sender, *params.values()) + + +class TestSignalSender: + """ + SignalSender tests. + """ + + def test_bind(self, signal, sender, binder): + """ + Tests signal is bound on init. + """ + cls = type(sender) + + assert isinstance(cls.on_signal, Signal) + assert isinstance(sender.on_signal, SignalBinder) + + assert signal == cls.on_signal + assert binder == sender.on_signal + + def test_send(self, params, sender): + """ + Tests sender is passed to signal + """ + with patch.object(Signal, 'send') as m: + sender.on_signal(*params.values()) + m.assert_called_once_with(sender, *params.values()) + + +class TestSignalReceiver: + """ + SignalReceiver tests. + """ + + def test_connect(self, signal, receiver): + """ + Tests receiver connects to signal automatically. + """ + assert receiver.on_signal in signal.receivers + + def test_send(self, params, sender, receiver): + """ + Tests receiver recives emitted singal. + """ + sender.on_signal(*params.values()) + receiver.on_signal.assert_called_once_with(sender, **params)