HiveBrain v1.2.0
Get Started
← Back to all entries
principlepythonModerate

The Observer design pattern in Python, with unit tests

Submitted by: @import:stackexchange-codereview··
0
Viewed 0 times
thewithobserverdesigntestspythonpatternunit

Problem

I'm continuing to work on the Head First Design Patterns book in an effort to become a more efficient and better Python programmer. Code review for the Strategy pattern in Chapter 1 is here with helpful explanations from the community.

I've tried to implement the Observer design pattern from chapter 2 below in Python. In the book all the examples are in Java. I'm sure there are more idiomatic Python things that I can do to this code. In the first example in this chapter we implement the Subject so that it pushes data out to the individual observers. In my code I wanted the Subject to notify the observers that new data was available, and have the observers pull only the data that each individual observer was interested in. I also tried to unit test instead of just printing output.

Some issues I don't know how to resolve:

-
The Observer abstract class has an abstract method update(), but two other non-abstract methods register_subject(self, subject) and remove_subject(self). So when I implement the concrete observers, it inherits these two methods. Is this a wrong way of using the Abstract Base Class?

-
I've recently started to unit test and have had some trouble formulating my tests (for this particular example I wrote the tests after the code, although I know it's better to write the tests first). A particular example: I test that the Observers are in fact registered with the Subject by looking to see that the observer instance is in the Subject._observer_list, but this list is hidden. Is it ok for the test to be looking there?

-
Any other comments or suggestions on the code?

Code

```
#!/usr/bin/env python

from abc import ABCMeta, abstractmethod
import unittest

"""Implementation of the Observer pattern from Head First Design Patters
(Chapter 2), using the pull method of passing data from the Subject to the
Observer(s). """

################################################################################
# Abstract classes
####################

Solution

A few random comments on pieces of the code:

class Subject:
    __metaclass__ = ABCMeta

    @abstractmethod
    def register_observer(observer):
        """Registers an observer with Subject."""
        pass

    @abstractmethod
    def remove_observer(observer):
        """Removes an observer from Subject."""
        pass

    @abstractmethod
    def notify_observers():
        """Notifies observers that Subject data has changed."""
        pass


Why aren't you defining these methods? It seems to me they'll be the same for all subclasses so you don't need to make them abstract and implement them there.

def register_observer(self, observer):
        """Registers an observer with WeatherData if the observer is not
        already registered."""
        try:
            if observer not in self._observer_list:
                self._observer_list.append(observer)
                observer.register_subject(self)
            else:
                raise ValueError


Don't throw generic exceptions that might get mistaken for something else. I'd raise Exception("Observer already subscribed to Subject!") or a custom exception class.

except ValueError:
            print "ERROR: Observer already subscribed to Subject!"


Don't print error messages, just throw exceptions.

raise ValueError


Certainly don't catch an exception right after throwing it and then throw it again.


The Observer abstract class has an abstract method update(), but two
other non-abstract methods register_subject(self, subject) and
remove_subject(self). So when I implement the concrete observers, it
inherits these two methods. Is this a wrong way of using the Abstract
Base Class?

That is a perfectly valid and common way of abstract base classes.


I've recently started to unit test and have had some trouble
formulating my tests (for this particular example I wrote the tests
after the code, although I know it's better to write the tests first).
A particular example: I test that the Observers are in fact registered
with the Subject by looking to see that the observer instance is in
the Subject._observer_list, but this list is hidden. Is it ok for the
test to be looking there?

No, you shouldn't test the internal state of the object. You should test the external actions of the object. Here is a sample of how you should test it:

class MyObserver(Observer):
    def __init__(self):
        self.updated = False

    def update(self):
        self.updated = True

def test_it():
    temperature_data = TemperatureData()
    observer = MyObserver()
    temperature_data.register_observer(observer)
    assert not observer.updated()
    temperature_data.set_measurements(2,4,4)
    assert observer.updated()


But the observer pattern is really not a great fit for python. It is not flexible and doesn't take advantage of python's features. The pattern I use is something like this:

class Signal(object):
    def __init__(self):
        self._handlers = []

    def connect(self, handler):
        self._handlers.append(handler)

    def fire(self, *args):
        for handler in self._handlers:
            handler(*args)


On the Subject I'd do this:

class TemperatureData:
    def __init__(self):
        self.changed = Signal()

    def set_temperaturedata(self, foo, bar):
        ...
        self.changed.fire(self)


Then to hook things up I'd say

display = TemperatureDisplay()
temperature_data.changed.connect(display.update_temperaturedata)


I think this model is simpler. It lets objects emit multiple signals that could be observed, a given object can observer multiple other objects. The TemperatureDisplay object doesn't even know where the data is coming from, it just magically shows up on update_temperaturedata making it easier to test that object in isolation.

Code Snippets

class Subject:
    __metaclass__ = ABCMeta

    @abstractmethod
    def register_observer(observer):
        """Registers an observer with Subject."""
        pass

    @abstractmethod
    def remove_observer(observer):
        """Removes an observer from Subject."""
        pass

    @abstractmethod
    def notify_observers():
        """Notifies observers that Subject data has changed."""
        pass
def register_observer(self, observer):
        """Registers an observer with WeatherData if the observer is not
        already registered."""
        try:
            if observer not in self._observer_list:
                self._observer_list.append(observer)
                observer.register_subject(self)
            else:
                raise ValueError
except ValueError:
            print "ERROR: Observer already subscribed to Subject!"
raise ValueError
class MyObserver(Observer):
    def __init__(self):
        self.updated = False

    def update(self):
        self.updated = True

def test_it():
    temperature_data = TemperatureData()
    observer = MyObserver()
    temperature_data.register_observer(observer)
    assert not observer.updated()
    temperature_data.set_measurements(2,4,4)
    assert observer.updated()

Context

StackExchange Code Review Q#20938, answer score: 16

Revisions (0)

No revisions yet.