Add signals module

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 <http://www.gnu.org/licenses/>.
+#
+
+
+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 "<signal ({})>".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 "<bound {} of {}>".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)

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 <http://www.gnu.org/licenses/>.
+#
+
+
+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)