import pytest from myutils.observable import NotObservableError, make_observable, bind, collect_return_values, unbind, unbind_all, \ has_listeners, get_listener_count, add_event_listener, ObservableEvent, remove_event_listener # Test fixtures class Data: def __init__(self): self.value: str = "initial" self.number: int = 1 def get_double(self): return self.number * 2 @pytest.fixture def data(): return Data() @pytest.fixture def observable_data(data): return make_observable(data) # Tests def test_i_can_make_an_object_observable(observable_data): assert hasattr(observable_data, '_listeners') assert isinstance(observable_data._listeners, dict) assert observable_data is observable_data def test_i_can_bind_a_callback_to_an_attribute(): data = Data() make_observable(data) callback = lambda old, new: None bind(data, 'number', callback) assert 'number' in data._listeners assert callback in data._listeners['number'] def test_i_can_receive_notification_when_attribute_changes(): data = Data() make_observable(data) called = [] bind(data, 'number', lambda old, new: called.append(True)) data.number = 5 assert len(called) == 1 def test_i_can_receive_old_and_new_values_in_callback(): data = Data() make_observable(data) values = [] bind(data, 'number', lambda old, new: values.append((old, new))) data.number = 5 data.number = 10 assert values == [(1, 5), (5, 10)] def test_i_can_bind_multiple_callbacks_to_same_attribute(): data = Data() make_observable(data) calls1 = [] calls2 = [] bind(data, 'number', lambda old, new: calls1.append(new)) bind(data, 'number', lambda old, new: calls2.append(new)) data.number = 5 assert calls1 == [5] assert calls2 == [5] def test_i_can_bind_callbacks_to_different_attributes(): data = Data() make_observable(data) number_calls = [] name_calls = [] bind(data, 'number', lambda old, new: number_calls.append(new)) bind(data, 'name', lambda old, new: name_calls.append(new)) data.number = 5 data.name = "test" assert number_calls == [5] assert name_calls == ["test"] def test_i_can_modify_non_observed_attributes_without_notification(): data = Data() make_observable(data) called = [] bind(data, 'number', lambda old, new: called.append(True)) data.other_attr = "value" assert len(called) == 0 assert data.other_attr == "value" def test_i_can_have_multiple_instances_with_independent_observers(): data1 = Data() data2 = Data() make_observable(data1) calls1 = [] bind(data1, 'number', lambda old, new: calls1.append(new)) data1.number = 5 data2.number = 10 # Only data1 should trigger callback assert calls1 == [5] assert data1.number == 5 assert data2.number == 10 def test_i_can_call_make_observable_multiple_times_safely(): data = Data() result1 = make_observable(data) result2 = make_observable(data) assert result1 is result2 assert hasattr(data, '_listeners') # Should still work normally called = [] bind(data, 'number', lambda old, new: called.append(new)) data.number = 5 assert len(called) == 1 def test_i_cannot_bind_before_making_observable(): data = Data() with pytest.raises(NotObservableError) as exc_info: bind(data, 'number', lambda old, new: None) assert "must be made observable" in str(exc_info.value).lower() def test_i_can_set_attribute_that_does_not_exist_yet(): data = Data() make_observable(data) values = [] bind(data, 'new_attr', lambda old, new: values.append((old, new))) data.new_attr = "first_value" assert values == [(None, "first_value")] assert data.new_attr == "first_value" def test_i_can_preserve_original_class_behavior(): data = Data() make_observable(data) # Test that original methods still work assert data.get_double() == 2 data.number = 5 assert data.get_double() == 10 # Test that isinstance still works assert isinstance(data, Data) def test_i_can_collect_the_updates(): data = Data() make_observable(data) bind(data, 'number', lambda old, new: new) bind(data, 'number', lambda old, new: new * 2) data.number = 5 assert collect_return_values(data) == [5, 10] # another time to make sure there is no side effect data.number = 10 assert collect_return_values(data) == [10, 20] def test_i_cannot_collect_updates_before_making_observable(): data = Data() with pytest.raises(NotObservableError) as exc_info: collect_return_values(data) assert "must be made observable" in str(exc_info.value).lower() def test_i_can_collect_none(): data = Data() make_observable(data) bind(data, 'number', lambda old, new: None) data.number = 5 assert collect_return_values(data) == [None] def test_i_can_unbind_a_callback(observable_data): """ Unbinding a callback should prevent it from being called. """ results = [] def callback(old, new): results.append((old, new)) bind(observable_data, "value", callback) observable_data.value = "first" assert len(results) == 1 unbind(observable_data, "value", callback) observable_data.value = "second" assert len(results) == 1 # Should not have increased def test_i_can_unbind_one_callback_without_affecting_others(observable_data): """ Unbinding one callback should not affect other callbacks on the same attribute. """ results1 = [] results2 = [] def callback1(old, new): results1.append((old, new)) def callback2(old, new): results2.append((old, new)) bind(observable_data, "value", callback1) bind(observable_data, "value", callback2) observable_data.value = "first" assert len(results1) == 1 assert len(results2) == 1 unbind(observable_data, "value", callback1) observable_data.value = "second" assert len(results1) == 1 # Should not have increased assert len(results2) == 2 # Should have increased def test_i_can_unbind_callback_from_one_attr_without_affecting_others(observable_data): """ Unbinding a callback from one attribute should not affect callbacks on other attributes. """ value_results = [] number_results = [] def value_callback(old, new): value_results.append((old, new)) def number_callback(old, new): number_results.append((old, new)) bind(observable_data, "value", value_callback) bind(observable_data, "number", number_callback) observable_data.value = "first" observable_data.number = 1 assert len(value_results) == 1 assert len(number_results) == 1 unbind(observable_data, "value", value_callback) observable_data.value = "second" observable_data.number = 2 assert len(value_results) == 1 # Should not have increased assert len(number_results) == 2 # Should have increased def test_unbind_with_non_existent_callback_does_not_raise(observable_data): """ Unbinding a callback that was never bound should fail silently. """ def callback(old, new): pass # Should not raise an exception unbind(observable_data, "value", callback) def test_unbind_with_non_existent_attribute_does_not_raise(observable_data): """ Unbinding from an attribute that has no listeners should fail silently. """ def callback(old, new): pass # Should not raise an exception unbind(observable_data, "non_existent", callback) def test_i_cannot_unbind_from_non_observable_object(data): """ Trying to unbind from a non-observable object should raise NotObservableError. """ def callback(old, new): pass with pytest.raises(NotObservableError, match="must be made observable"): unbind(data, "value", callback) def test_unbind_same_callback_multiple_times_is_safe(observable_data): """ Unbinding the same callback multiple times should be safe. """ def callback(old, new): pass bind(observable_data, "value", callback) unbind(observable_data, "value", callback) unbind(observable_data, "value", callback) # Should not raise unbind(observable_data, "value", callback) # Should not raise # ============================================================================ # unbind_all() tests # ============================================================================ def test_i_can_unbind_all_callbacks_from_one_attribute(observable_data): """ unbind_all() with an attribute should remove all callbacks from that attribute. """ results1 = [] results2 = [] def callback1(old, new): results1.append((old, new)) def callback2(old, new): results2.append((old, new)) bind(observable_data, "value", callback1) bind(observable_data, "value", callback2) observable_data.value = "first" assert len(results1) == 1 assert len(results2) == 1 unbind_all(observable_data, "value") observable_data.value = "second" assert len(results1) == 1 # Should not have increased assert len(results2) == 1 # Should not have increased def test_unbind_all_on_one_attr_does_not_affect_others(observable_data): """ unbind_all() on one attribute should not affect listeners on other attributes. """ value_results = [] count_results = [] def value_callback(old, new): value_results.append((old, new)) def count_callback(old, new): count_results.append((old, new)) bind(observable_data, "value", value_callback) bind(observable_data, "count", count_callback) unbind_all(observable_data, "value") observable_data.value = "first" observable_data.count = 1 assert len(value_results) == 0 # Should not have increased assert len(count_results) == 1 # Should have increased def test_i_can_unbind_all_callbacks_from_all_attributes(observable_data): """ unbind_all() without an attribute should remove all callbacks from all attributes. """ value_results = [] count_results = [] def value_callback(old, new): value_results.append((old, new)) def count_callback(old, new): count_results.append((old, new)) bind(observable_data, "value", value_callback) bind(observable_data, "count", count_callback) unbind_all(observable_data) observable_data.value = "first" observable_data.count = 1 assert len(value_results) == 0 assert len(count_results) == 0 def test_i_cannot_unbind_all_from_non_observable_object(data): """ Trying to unbind_all from a non-observable object should raise NotObservableError. """ with pytest.raises(NotObservableError, match="must be made observable"): unbind_all(data, "value") def test_unbind_all_with_non_existent_attribute_is_safe(observable_data): """ unbind_all() with a non-existent attribute should not raise an error. """ # Should not raise unbind_all(observable_data, "non_existent") # ============================================================================ # has_listeners() tests # ============================================================================ def test_has_listeners_returns_true_when_listeners_exist(observable_data): """ has_listeners() should return True when there are listeners. """ def callback(old, new): pass bind(observable_data, "value", callback) assert has_listeners(observable_data, "value") is True def test_has_listeners_returns_false_when_no_listeners(observable_data): """ has_listeners() should return False when there are no listeners. """ assert has_listeners(observable_data, "value") is False def test_has_listeners_returns_false_after_unbind(observable_data): """ has_listeners() should return False after all listeners are removed. """ def callback(old, new): pass bind(observable_data, "value", callback) assert has_listeners(observable_data, "value") is True unbind(observable_data, "value", callback) assert has_listeners(observable_data, "value") is False def test_i_cannot_check_listeners_on_non_observable_object(data): """ Trying to check listeners on a non-observable object should raise NotObservableError. """ with pytest.raises(NotObservableError, match="must be made observable"): has_listeners(data, "value") # ============================================================================ # get_listener_count() tests # ============================================================================ def test_get_listener_count_returns_zero_when_no_listeners(observable_data): """ get_listener_count() should return 0 when there are no listeners. """ assert get_listener_count(observable_data, "value") == 0 def test_get_listener_count_returns_correct_count(observable_data): """ get_listener_count() should return the correct number of listeners. """ def callback1(old, new): pass def callback2(old, new): pass def callback3(old, new): pass bind(observable_data, "value", callback1) assert get_listener_count(observable_data, "value") == 1 bind(observable_data, "value", callback2) assert get_listener_count(observable_data, "value") == 2 bind(observable_data, "value", callback3) assert get_listener_count(observable_data, "value") == 3 def test_get_listener_count_decreases_after_unbind(observable_data): """ get_listener_count() should decrease after unbinding a callback. """ def callback1(old, new): pass def callback2(old, new): pass bind(observable_data, "value", callback1) bind(observable_data, "value", callback2) assert get_listener_count(observable_data, "value") == 2 unbind(observable_data, "value", callback1) assert get_listener_count(observable_data, "value") == 1 unbind(observable_data, "value", callback2) assert get_listener_count(observable_data, "value") == 0 def test_i_cannot_get_listener_count_on_non_observable_object(data): """ Trying to get listener count on a non-observable object should raise NotObservableError. """ with pytest.raises(NotObservableError, match="must be made observable"): get_listener_count(data, "value") # ============================================================================ # add_event_listener() tests # ============================================================================ def test_i_can_add_an_event_listener(observable_data): """ add_event_listener() should correctly add a listener for a specific event and attribute. """ def callback(attr, old, new, results): pass add_event_listener( ObservableEvent.AFTER_PROPERTY_CHANGE, observable_data, "value", callback ) # Check that the listener has been added event_name = ObservableEvent.AFTER_PROPERTY_CHANGE.value assert event_name in observable_data._listeners assert "value" in observable_data._listeners[event_name] assert callback in observable_data._listeners[event_name]["value"] def test_i_can_add_multiple_event_listeners_for_same_attribute(observable_data): """ Multiple event listeners can be added to the same attribute and event. """ def callback1(attr, old, new, results): pass def callback2(attr, old, new, results): pass def callback3(attr, old, new, results): pass event = ObservableEvent.AFTER_PROPERTY_CHANGE add_event_listener(event, observable_data, "value", callback1) add_event_listener(event, observable_data, "value", callback2) add_event_listener(event, observable_data, "value", callback3) # Check that all listeners have been added event_name = event.value assert len(observable_data._listeners[event_name]["value"]) == 3 assert callback1 in observable_data._listeners[event_name]["value"] assert callback2 in observable_data._listeners[event_name]["value"] assert callback3 in observable_data._listeners[event_name]["value"] def test_i_can_handle_duplicate_event_listeners(observable_data): """ The same callback can be added multiple times and should appear multiple times. """ def callback(attr, old, new, results): pass event = ObservableEvent.AFTER_PROPERTY_CHANGE add_event_listener(event, observable_data, "value", callback) add_event_listener(event, observable_data, "value", callback) add_event_listener(event, observable_data, "value", callback) # Check that the same callback appears 3 times event_name = event.value assert len(observable_data._listeners[event_name]["value"]) == 3 assert observable_data._listeners[event_name]["value"].count(callback) == 3 def test_i_cannot_add_listener_to_non_observable_object(data): """ Trying to add an event listener to a non-observable object should raise NotObservableError. """ def callback(attr, old, new, results): pass with pytest.raises(NotObservableError, match="must be made observable"): add_event_listener(ObservableEvent.AFTER_PROPERTY_CHANGE, data, "value", callback) # ============================================================================ # remove_event_listener() tests # ============================================================================ def test_i_can_remove_an_event_listener(observable_data): """ remove_event_listener() should correctly remove a previously added listener. """ def callback(attr, old, new, results): pass event = ObservableEvent.AFTER_PROPERTY_CHANGE # Add the listener add_event_listener(event, observable_data, "value", callback) event_name = event.value assert callback in observable_data._listeners[event_name]["value"] # Remove the listener remove_event_listener(event, observable_data, "value", callback) # Check that the listener has been removed and structure cleaned up assert event_name not in observable_data._listeners def test_i_can_remove_last_event_listener_and_clean_up_structure(observable_data): """ Removing the last listener should clean up the internal data structures. """ def callback(attr, old, new, results): pass event = ObservableEvent.AFTER_PROPERTY_CHANGE event_name = event.value add_event_listener(event, observable_data, "value", callback) # Verify it was added assert event_name in observable_data._listeners assert "value" in observable_data._listeners[event_name] # Remove it remove_event_listener(event, observable_data, "value", callback) # Verify cleanup: event should be removed from _listeners assert event_name not in observable_data._listeners def test_i_cannot_remove_listener_from_non_observable_object(data): """ Trying to remove an event listener from a non-observable object should raise NotObservableError. """ def callback(attr, old, new, results): pass with pytest.raises(NotObservableError, match="must be made observable"): remove_event_listener(ObservableEvent.AFTER_PROPERTY_CHANGE, data, "value", callback) def test_i_cannot_remove_non_existent_event_listener(observable_data): """ Trying to remove a non-existent listener should fail silently (no exception). """ def callback(attr, old, new, results): pass event = ObservableEvent.AFTER_PROPERTY_CHANGE # Should not raise an exception remove_event_listener(event, observable_data, "value", callback) def test_remove_event_listener_does_not_affect_others(observable_data): """ Removing one listener should not affect other listeners on the same attribute. """ def callback1(attr, old, new, results): pass def callback2(attr, old, new, results): pass def callback3(attr, old, new, results): pass event = ObservableEvent.AFTER_PROPERTY_CHANGE event_name = event.value # Add three listeners add_event_listener(event, observable_data, "value", callback1) add_event_listener(event, observable_data, "value", callback2) add_event_listener(event, observable_data, "value", callback3) assert len(observable_data._listeners[event_name]["value"]) == 3 # Remove only the second one remove_event_listener(event, observable_data, "value", callback2) # Check that only callback2 was removed assert len(observable_data._listeners[event_name]["value"]) == 2 assert callback1 in observable_data._listeners[event_name]["value"] assert callback2 not in observable_data._listeners[event_name]["value"] assert callback3 in observable_data._listeners[event_name]["value"] # ============================================================================ # Integration tests for add/remove event listeners # ============================================================================ def test_i_can_add_and_remove_event_listeners_sequentially(observable_data): """ Multiple add/remove cycles should work correctly. """ def callback(attr, old, new, results): pass event = ObservableEvent.AFTER_PROPERTY_CHANGE event_name = event.value # Cycle 1: Add and remove add_event_listener(event, observable_data, "value", callback) assert event_name in observable_data._listeners remove_event_listener(event, observable_data, "value", callback) assert event_name not in observable_data._listeners # Cycle 2: Add and remove again add_event_listener(event, observable_data, "value", callback) assert event_name in observable_data._listeners remove_event_listener(event, observable_data, "value", callback) assert event_name not in observable_data._listeners def test_listener_removal_on_multiple_event_types(observable_data): """ Listeners for different attributes should be independently manageable. """ def callback1(attr, old, new, results): pass def callback2(attr, old, new, results): pass event = ObservableEvent.AFTER_PROPERTY_CHANGE event_name = event.value # Add listeners to different attributes add_event_listener(event, observable_data, "value", callback1) add_event_listener(event, observable_data, "number", callback2) assert "value" in observable_data._listeners[event_name] assert "number" in observable_data._listeners[event_name] # Remove listener from "value" remove_event_listener(event, observable_data, "value", callback1) # Check that "value" is removed but "number" remains assert "value" not in observable_data._listeners[event_name] assert "number" in observable_data._listeners[event_name] assert callback2 in observable_data._listeners[event_name]["number"] def test_i_can_remove_all_event_listener_when_attr_name_is_none(observable_data): """ When unbind_all() is called with attr_name=None, all listeners for all attributes should be removed from the observable object. """ def callback1(attr, old, new, results): pass def callback2(attr, old, new, results): pass event = ObservableEvent.AFTER_PROPERTY_CHANGE event_name = event.value # Add listeners to different attributes add_event_listener(event, observable_data, "value", callback1) add_event_listener(event, observable_data, "value", callback2) add_event_listener(event, observable_data, "number", callback1) add_event_listener(event, observable_data, "number", callback2) remove_event_listener(event, observable_data, None, callback1) callbacks = observable_data._listeners[event_name]["value"] + observable_data._listeners[event_name]["number"] assert callback1 not in callbacks def test_i_can_remove_all_event_listener_when_callback_is_none(observable_data): """ When unbind_all() is called with attr_name=None, all listeners for all attributes should be removed from the observable object. """ def callback1(attr, old, new, results): pass def callback2(attr, old, new, results): pass event = ObservableEvent.AFTER_PROPERTY_CHANGE event_name = event.value # Add listeners to different attributes add_event_listener(event, observable_data, "value", callback1) add_event_listener(event, observable_data, "value", callback2) add_event_listener(event, observable_data, "number", callback1) add_event_listener(event, observable_data, "number", callback2) remove_event_listener(event, observable_data, "value", None) assert "value" not in observable_data._listeners[event_name] def test_i_can_remove_all_event_listener_when_attr_name_callback_are_none(observable_data): """ When unbind_all() is called with attr_name=None, all listeners for all attributes should be removed from the observable object. """ def callback1(attr, old, new, results): pass def callback2(attr, old, new, results): pass event = ObservableEvent.AFTER_PROPERTY_CHANGE event_name = event.value # Add listeners to different attributes add_event_listener(event, observable_data, "value", callback1) add_event_listener(event, observable_data, "value", callback2) add_event_listener(event, observable_data, "number", callback1) add_event_listener(event, observable_data, "number", callback2) remove_event_listener(event, observable_data, None, None) assert "value" not in observable_data._listeners[event_name] assert "number" not in observable_data._listeners[event_name] # ============================================================================ # Integration tests # ============================================================================ def test_multiple_bind_unbind_cycles(observable_data): """ Multiple bind/unbind cycles should work correctly. """ results = [] def callback(old, new): results.append((old, new)) # Cycle 1 bind(observable_data, "value", callback) observable_data.value = "first" assert len(results) == 1 unbind(observable_data, "value", callback) observable_data.value = "second" assert len(results) == 1 # Cycle 2 bind(observable_data, "value", callback) observable_data.value = "third" assert len(results) == 2 unbind(observable_data, "value", callback) observable_data.value = "fourth" assert len(results) == 2 # ============================================================================ # After property changed event # ============================================================================ def test_i_can_receive_changes_in_after_property_change_event(observable_data): """Default case : after_property_change event should be fired when property changes.""" data = Data() make_observable(data) on_change_results = [] on_after_change_results = [] def on_change(old, new): on_change_results.append((old, new)) return new + 1 def on_after_change(attr, old, new, results): on_after_change_results.append((attr, old, new, results)) bind(data, 'number', on_change) add_event_listener(ObservableEvent.AFTER_PROPERTY_CHANGE, data, "number", on_after_change) data.number = 5 data.number = 10 assert on_change_results == [(1, 5), (5, 10)] assert on_after_change_results == [("number", 1, 5, [6]), ("number", 5, 10, [11])] def test_i_can_receive_changes_when_multiple_properties_are_changed(observable_data): """When multiple properties are changed, after_property_change event fires when all properties change.""" data = Data() make_observable(data) def on_change_1(old, new): return new def on_change_2(old, new): return new + 1 on_after_change_results = [] def on_after_change(attr, old, new, results): on_after_change_results.append((attr, old, new, results)) bind(data, 'number', on_change_1) bind(data, 'number', on_change_2) add_event_listener(ObservableEvent.AFTER_PROPERTY_CHANGE, data, "number", on_after_change) data.number = 5 data.number = 10 assert on_after_change_results == [("number", 1, 5, [5, 6]), ("number", 5, 10, [10, 11])] def test_i_can_receive_changes_in_after_property_change_event_when_declared_first(observable_data): """after_property_change event should be fired even if defined before the bindings.""" data = Data() make_observable(data) def on_change(old, new): return new + 1 on_after_change_results = [] def on_after_change(attr, old, new, results): on_after_change_results.append((attr, old, new, results)) add_event_listener(ObservableEvent.AFTER_PROPERTY_CHANGE, data, "number", on_after_change) bind(data, 'number', on_change) data.number = 5 data.number = 10 assert on_after_change_results == [("number", 1, 5, [6]), ("number", 5, 10, [11])] def test_i_can_receive_changes_in_after_property_change_event_for_requested_attribute(observable_data): """after_property_change event should be fired even if defined before the bindings.""" data = Data() make_observable(data) def on_change_1(old, new): return new def on_change_2(old, new): return str(new) + "_1" on_after_change_results = [] def on_after_change(attr, old, new, results): on_after_change_results.append((attr, old, new, results)) add_event_listener(ObservableEvent.AFTER_PROPERTY_CHANGE, data, "number", on_after_change) bind(data, 'number', on_change_1) bind(data, 'number', on_change_2) bind(data, 'value', on_change_1) data.number = 5 data.value = "new value" assert on_after_change_results == [("number", 1, 5, [5, "5_1"])] def test_i_can_receive_changes_in_after_property_change_event_for_all_attributes(observable_data): """after_property_change event should be fired even if defined before the bindings.""" data = Data() make_observable(data) def on_change_1(old, new): return new def on_change_2(old, new): return str(new) + "_1" on_after_change_results = [] def on_after_change(attr, old, new, results): on_after_change_results.append((attr, old, new, results)) add_event_listener(ObservableEvent.AFTER_PROPERTY_CHANGE, data, "", on_after_change) bind(data, 'number', on_change_1) bind(data, 'number', on_change_2) bind(data, 'value', on_change_1) data.number = 5 data.value = "new value" assert on_after_change_results == [("number", 1, 5, [5, "5_1"]), ("value", "initial", "new value", ["new value"])]