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 pythonfrom abc import ABCMeta, abstractmethodimport unittestImplementation of the Observer pattern from Head First Design Patters(Chapter 2), using the pull method of passing data from the Subject to theObserver(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.        passclass 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 = Noneclass 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 weatherclass 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()OutputCurrent conditions: 80.0 F degrees and 65.0 % humidityImproving weather on the way!Avg/Max/Min temperature = 80.0/80.0/80.0Current conditions: 82.0 F degrees and 70.0 % humiditywatch out for cooler, rainy weatherAvg/Max/Min temperature = 81.0/82.0/80.0Current conditions: 87.0 F degrees and 90.0 % humidityMore of the sameAvg/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.001sOK
The Observer design pattern in Python in a more pythonic way (plus unit testing best practices)
python;design patterns;unit testing
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.        passWhy 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 ValueErrorDon'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 ValueErrorCertainly 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 = Truedef 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 saydisplay = 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.