1290 lines
37 KiB
Python
1290 lines
37 KiB
Python
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, \
|
|
ObservableResultCollector
|
|
|
|
|
|
# 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"])]
|
|
|
|
|
|
# ============================================================================
|
|
# ObservableResultCollector tests
|
|
# ============================================================================
|
|
|
|
def test_i_can_create_collector_with_list_of_observables():
|
|
"""
|
|
ObservableResultCollector should accept a list of observable objects.
|
|
"""
|
|
data1 = make_observable(Data())
|
|
data2 = make_observable(Data())
|
|
|
|
collector = ObservableResultCollector([data1, data2])
|
|
|
|
assert collector.observables == [data1, data2]
|
|
assert collector.attr_name == ""
|
|
assert collector.event == ObservableEvent.AFTER_PROPERTY_CHANGE
|
|
assert collector.results == []
|
|
|
|
|
|
def test_i_can_create_collector_with_single_observable():
|
|
"""
|
|
ObservableResultCollector should automatically convert a single object to a list.
|
|
"""
|
|
data = make_observable(Data())
|
|
|
|
collector = ObservableResultCollector(data)
|
|
|
|
assert collector.observables == [data]
|
|
assert isinstance(collector.observables, list)
|
|
|
|
|
|
def test_i_can_collect_results_from_single_observable():
|
|
"""
|
|
Collector should collect results from a callback bound to an observable.
|
|
"""
|
|
data = make_observable(Data())
|
|
|
|
def callback(old, new):
|
|
return f"Changed from {old} to {new}"
|
|
|
|
bind(data, "value", callback)
|
|
|
|
with ObservableResultCollector(data) as collector:
|
|
data.value = "world"
|
|
|
|
assert collector.results == ["Changed from initial to world"]
|
|
|
|
|
|
def test_i_can_collect_results_from_multiple_callbacks():
|
|
"""
|
|
Collector should collect results from multiple callbacks on the same attribute.
|
|
"""
|
|
data = make_observable(Data())
|
|
|
|
def callback1(old, new):
|
|
return f"callback1: {new}"
|
|
|
|
def callback2(old, new):
|
|
return f"callback2: {new}"
|
|
|
|
bind(data, "value", callback1)
|
|
bind(data, "value", callback2)
|
|
|
|
with ObservableResultCollector(data) as collector:
|
|
data.value = "test"
|
|
|
|
assert collector.results == ["callback1: test", "callback2: test"]
|
|
|
|
|
|
def test_i_can_collect_results_from_multiple_observables():
|
|
"""
|
|
Collector should collect results from multiple observable objects.
|
|
"""
|
|
data1 = make_observable(Data())
|
|
data2 = make_observable(Data())
|
|
|
|
def callback(old, new):
|
|
return f"Changed to {new}"
|
|
|
|
bind(data1, "value", callback)
|
|
bind(data2, "value", callback)
|
|
|
|
with ObservableResultCollector([data1, data2]) as collector:
|
|
data1.value = "first"
|
|
data2.value = "second"
|
|
|
|
assert collector.results == ["Changed to first", "Changed to second"]
|
|
|
|
|
|
def test_i_can_collect_results_for_specific_attribute():
|
|
"""
|
|
Collector should only collect results for a specific attribute when specified.
|
|
"""
|
|
data = make_observable(Data())
|
|
|
|
def value_callback(old, new):
|
|
return f"value: {new}"
|
|
|
|
def number_callback(old, new):
|
|
return f"number: {new}"
|
|
|
|
bind(data, "value", value_callback)
|
|
bind(data, "number", number_callback)
|
|
|
|
with ObservableResultCollector(data, attr_name="value") as collector:
|
|
data.value = "test"
|
|
data.number = 42
|
|
|
|
# Should only collect results from value changes
|
|
assert collector.results == ["value: test"]
|
|
|
|
|
|
def test_i_can_collect_results_for_all_attributes():
|
|
"""
|
|
Collector with attr_name="" should collect results from all attributes.
|
|
"""
|
|
data = make_observable(Data())
|
|
|
|
def value_callback(old, new):
|
|
return f"value: {new}"
|
|
|
|
def number_callback(old, new):
|
|
return f"number: {new}"
|
|
|
|
bind(data, "value", value_callback)
|
|
bind(data, "number", number_callback)
|
|
|
|
with ObservableResultCollector(data, attr_name="") as collector:
|
|
data.value = "test"
|
|
data.number = 42
|
|
|
|
# Should collect results from all attributes
|
|
assert collector.results == ["value: test", "number: 42"]
|
|
|
|
|
|
def test_i_can_verify_listeners_are_removed_after_exit():
|
|
"""
|
|
Listeners should be automatically removed when exiting the context manager.
|
|
"""
|
|
data = make_observable(Data())
|
|
|
|
def callback(old, new):
|
|
return f"Changed to {new}"
|
|
|
|
bind(data, "value", callback)
|
|
|
|
event_name = ObservableEvent.AFTER_PROPERTY_CHANGE.value
|
|
|
|
# Before entering context, no event listeners
|
|
assert event_name not in data._listeners
|
|
|
|
with ObservableResultCollector(data) as collector:
|
|
# Inside context, event listener should be added
|
|
assert event_name in data._listeners
|
|
assert "" in data._listeners[event_name]
|
|
|
|
data.value = "test"
|
|
|
|
# After exiting context, event listener should be removed
|
|
assert event_name not in data._listeners
|
|
assert collector.results == ["Changed to test"]
|
|
|
|
|
|
def test_i_can_verify_listeners_are_removed_on_exception():
|
|
"""
|
|
Listeners should be removed even when an exception occurs in the with block.
|
|
"""
|
|
data = make_observable(Data())
|
|
|
|
def callback(old, new):
|
|
return f"Changed to {new}"
|
|
|
|
bind(data, "value", callback)
|
|
|
|
event_name = ObservableEvent.AFTER_PROPERTY_CHANGE.value
|
|
|
|
try:
|
|
with ObservableResultCollector(data) as collector:
|
|
data.value = "test"
|
|
raise ValueError("Test exception")
|
|
except ValueError:
|
|
pass
|
|
|
|
# Listener should be removed despite exception
|
|
assert event_name not in data._listeners
|
|
|
|
|
|
def test_i_can_access_results_through_collector_object():
|
|
"""
|
|
The collector object returned by __enter__ should provide access to results.
|
|
"""
|
|
data = make_observable(Data())
|
|
|
|
def callback(old, new):
|
|
return new * 2
|
|
|
|
bind(data, "number", callback)
|
|
|
|
with ObservableResultCollector(data) as collector:
|
|
data.number = 5
|
|
# Results should be accessible during the with block
|
|
assert collector.results == [10]
|
|
|
|
data.number = 10
|
|
assert collector.results == [10, 20]
|
|
|
|
|
|
def test_i_can_collect_empty_results_when_no_callbacks():
|
|
"""
|
|
Collector should have empty results if no callbacks are bound.
|
|
"""
|
|
data = make_observable(Data())
|
|
|
|
with ObservableResultCollector(data) as collector:
|
|
data.value = "test"
|
|
|
|
# No callbacks were bound, so results should be empty
|
|
assert collector.results == []
|
|
|
|
|
|
def test_i_can_collect_none_values():
|
|
"""
|
|
Collector should collect None values returned by callbacks.
|
|
"""
|
|
data = make_observable(Data())
|
|
|
|
def callback1(old, new):
|
|
return None
|
|
|
|
def callback2(old, new):
|
|
return "not none"
|
|
|
|
bind(data, "value", callback1)
|
|
bind(data, "value", callback2)
|
|
|
|
with ObservableResultCollector(data) as collector:
|
|
data.value = "test"
|
|
|
|
assert collector.results == [None, "not none"]
|