The Observer design pattern in Python, with unit tests

Question

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
################################################################################
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
class Observer:
 __metaclass__ = ABCMeta
 @abstractmethod
 def update():
 """Observer updates by pulling data from Subject."""
 pass
 def register_subject(self, subject):
 """Observer saves reference to Subject."""
 self.subject = subject
 def remove_subject(self):
 """Observer replaces Subject reference to None."""
 self.subject = None
class DisplayElement:
 __metaclass__ = ABCMeta
 @abstractmethod
 def display():
 """DisplayElement displays instance data."""
 pass
################################################################################
# Concrete Subject class
################################################################################
class WeatherData(Subject):
 def __init__(self):
 self._observer_list = []
 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
 def remove_observer(self, observer):
 """Removes an observer from WeatherData if the observer is currently
 subscribed to WeatherData."""
 try:
 if observer in self._observer_list:
 observer.remove_subject()
 self._observer_list.remove(observer)
 else:
 raise ValueError
 except ValueError:
 print "ERROR: Observer currently not subscribed to Subject!"
 raise ValueError
 def notify_observers(self):
 """Notifies subscribed observers of change in WeatherData data."""
 for observer in self._observer_list:
 observer.update()
 def set_measurements(self, temperature, humidity, pressure):
 """Function used to simulate weather station device and send out new
 data."""
 self.temperature = temperature
 self.humidity = humidity
 self.pressure = pressure
 self.notify_observers()
################################################################################
# Concrete Observer/DisplayElement classes
################################################################################
class CurrentConditionDisplay(Observer, DisplayElement):
 """Returns the current temperature and humidity."""
 def update(self):
 self.temperature = self.subject.temperature
 self.humidity = self.subject.humidity
 self.display()
 def display(self):
 print "Current conditions: %.1f F degrees and %.1f %% humidity" %\
 (self.temperature, self.humidity)
class ForecastDisplay(Observer, DisplayElement):
 """Returns the a forcast of the weather based on pressure readings."""
 def __init__(self):
 self.current_pressure = 29.92
 def update(self):
 self.last_pressure = self.current_pressure
 self.current_pressure = self.subject.pressure
 self.display()
 def display(self):
 if self.current_pressure > self.last_pressure:
 print "Improving weather on the way!"
 elif self.current_pressure == self.last_pressure:
 print "More of the same"
 elif self.current_pressure < self.last_pressure:
 print "watch out for cooler, rainy weather"
class StatisticsDisplay(Observer, DisplayElement):
 """Returns the temperature statistics for all readings."""
 def __init__(self):
 self._min_temp = 200
 self._max_temp = 0
 self._temp_sum = 0
 self._num_readings = 0
 def update(self):
 if self.subject.temperature > self._max_temp:
 self._max_temp = self.subject.temperature
 if self.subject.temperature < self._min_temp:
 self._min_temp = self.subject.temperature
 self._temp_sum += self.subject.temperature
 self._num_readings += 1
 self.display()
 def display(self):
 print "Avg/Max/Min temperature = %.1f/%.1f/%.1f" %\
 (float(self._temp_sum/self._num_readings),
 self._max_temp, self._min_temp)
################################################################################
# Unit tests
################################################################################
class TestWeatherData(unittest.TestCase):
 def setUp(self):
 self.weather_data = WeatherData()
 self.current_condition_display = CurrentConditionDisplay()
 self.forecast_display = ForecastDisplay()
 self.statistics_display = StatisticsDisplay()
 def test_register_observers(self):
 # Register observers with subject.
 self.weather_data.register_observer(self.current_condition_display)
 self.weather_data.register_observer(self.forecast_display)
 self.weather_data.register_observer(self.statistics_display)
 # Check registered observers in _observer_list.
 self.assertIn(self.current_condition_display,
 self.weather_data._observer_list)
 self.assertIn(self.forecast_display,
 self.weather_data._observer_list)
 self.assertIn(self.statistics_display,
 self.weather_data._observer_list)
 def test_register_remove_forcast_display(self):
 # Register observer with subject.
 self.weather_data.register_observer(self.forecast_display)
 # Check registered observer in _observer_list.
 self.assertIn(self.forecast_display,
 self.weather_data._observer_list)
 # Remove observer from subject.
 self.weather_data.remove_observer(self.forecast_display)
 # Check observer not in _observer_list.
 self.assertNotIn(self.forecast_display,
 self.weather_data._observer_list)
 def test_remove_nonexistant_display(self):
 # Check removing an observer already not in _observer_list throws a
 # ValueError.
 with self.assertRaises(ValueError):
 self.weather_data.remove_observer(self.forecast_display)
 def test_no_register_display_twice_display(self):
 # Check that adding an observer twice to in _observer_list throws a
 # ValueError.
 with self.assertRaises(ValueError):
 self.weather_data.register_observer(self.forecast_display)
 self.weather_data.register_observer(self.forecast_display)
class TestDisplays(unittest.TestCase):
 def setUp(self):
 self.weather_data = WeatherData()
 self.current_condition_display = CurrentConditionDisplay()
 self.forecast_display = ForecastDisplay()
 self.statistics_display = StatisticsDisplay()
 self.weather_data.register_observer(self.current_condition_display)
 self.weather_data.register_observer(self.forecast_display)
 self.weather_data.register_observer(self.statistics_display)
 def test_set_measurements(self):
 # Setting measurements and making sure they are passed correctly.
 self.weather_data.set_measurements(80, 65, 30.4)
 self.assertEqual(self.current_condition_display.temperature, 80)
 self.assertEqual(self.current_condition_display.humidity, 65)
 self.assertEqual(self.forecast_display.current_pressure, 30.4)
 self.weather_data.set_measurements(82, 70, 29.2)
 self.assertEqual(self.current_condition_display.temperature, 82)
 self.assertEqual(self.current_condition_display.humidity, 70)
 self.assertEqual(self.forecast_display.current_pressure, 29.2)
 self.weather_data.set_measurements(87, 90, 29.2)
 self.assertEqual(self.current_condition_display.temperature, 87)
 self.assertEqual(self.current_condition_display.humidity, 90)
 self.assertEqual(self.forecast_display.current_pressure, 29.2)
if __name__ == '__main__':
 unittest.main()

Output

Current conditions: 80.0 F degrees and 65.0 % humidity
Improving weather on the way!
Avg/Max/Min temperature = 80.0/80.0/80.0
Current conditions: 82.0 F degrees and 70.0 % humidity
watch out for cooler, rainy weather
Avg/Max/Min temperature = 81.0/82.0/80.0
Current conditions: 87.0 F degrees and 90.0 % humidity
More of the same
Avg/Max/Min temperature = 83.0/87.0/80.0
.ERROR: Observer already subscribed to Subject!
...ERROR: Observer currently not subscribed to Subject!
.
----------------------------------------------------------------------
Ran 5 tests in 0.001s
OK

Answer 1

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.