Added unbind, unbind_all, has_listeners get_listener_count
This commit is contained in:
@@ -1,207 +1,607 @@
|
||||
import pytest
|
||||
|
||||
from myutils.observable import NotObservableError, make_observable, bind, collect_return_values
|
||||
from myutils.observable import NotObservableError, make_observable, bind, collect_return_values, unbind, unbind_all, \
|
||||
has_listeners, get_listener_count
|
||||
|
||||
|
||||
# Test fixtures
|
||||
class Demo:
|
||||
class Data:
|
||||
def __init__(self):
|
||||
self.number = 1
|
||||
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():
|
||||
demo = Demo()
|
||||
result = make_observable(demo)
|
||||
|
||||
assert hasattr(demo, '_listeners')
|
||||
assert isinstance(demo._listeners, dict)
|
||||
assert result is demo
|
||||
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():
|
||||
demo = Demo()
|
||||
make_observable(demo)
|
||||
data = Data()
|
||||
make_observable(data)
|
||||
|
||||
callback = lambda old, new: None
|
||||
bind(demo, 'number', callback)
|
||||
bind(data, 'number', callback)
|
||||
|
||||
assert 'number' in demo._listeners
|
||||
assert callback in demo._listeners['number']
|
||||
assert 'number' in data._listeners
|
||||
assert callback in data._listeners['number']
|
||||
|
||||
|
||||
def test_i_can_receive_notification_when_attribute_changes():
|
||||
demo = Demo()
|
||||
make_observable(demo)
|
||||
data = Data()
|
||||
make_observable(data)
|
||||
|
||||
called = []
|
||||
bind(demo, 'number', lambda old, new: called.append(True))
|
||||
bind(data, 'number', lambda old, new: called.append(True))
|
||||
|
||||
demo.number = 5
|
||||
data.number = 5
|
||||
|
||||
assert len(called) == 1
|
||||
|
||||
|
||||
def test_i_can_receive_old_and_new_values_in_callback():
|
||||
demo = Demo()
|
||||
make_observable(demo)
|
||||
data = Data()
|
||||
make_observable(data)
|
||||
|
||||
values = []
|
||||
bind(demo, 'number', lambda old, new: values.append((old, new)))
|
||||
bind(data, 'number', lambda old, new: values.append((old, new)))
|
||||
|
||||
demo.number = 5
|
||||
demo.number = 10
|
||||
data.number = 5
|
||||
data.number = 10
|
||||
|
||||
assert values == [(1, 5), (5, 10)]
|
||||
|
||||
|
||||
def test_i_can_bind_multiple_callbacks_to_same_attribute():
|
||||
demo = Demo()
|
||||
make_observable(demo)
|
||||
data = Data()
|
||||
make_observable(data)
|
||||
|
||||
calls1 = []
|
||||
calls2 = []
|
||||
bind(demo, 'number', lambda old, new: calls1.append(new))
|
||||
bind(demo, 'number', lambda old, new: calls2.append(new))
|
||||
bind(data, 'number', lambda old, new: calls1.append(new))
|
||||
bind(data, 'number', lambda old, new: calls2.append(new))
|
||||
|
||||
demo.number = 5
|
||||
data.number = 5
|
||||
|
||||
assert calls1 == [5]
|
||||
assert calls2 == [5]
|
||||
|
||||
|
||||
def test_i_can_bind_callbacks_to_different_attributes():
|
||||
demo = Demo()
|
||||
make_observable(demo)
|
||||
data = Data()
|
||||
make_observable(data)
|
||||
|
||||
number_calls = []
|
||||
name_calls = []
|
||||
bind(demo, 'number', lambda old, new: number_calls.append(new))
|
||||
bind(demo, 'name', lambda old, new: name_calls.append(new))
|
||||
bind(data, 'number', lambda old, new: number_calls.append(new))
|
||||
bind(data, 'name', lambda old, new: name_calls.append(new))
|
||||
|
||||
demo.number = 5
|
||||
demo.name = "test"
|
||||
data.number = 5
|
||||
data.name = "test"
|
||||
|
||||
assert number_calls == [5]
|
||||
assert name_calls == ["test"]
|
||||
|
||||
|
||||
def test_i_can_modify_non_observed_attributes_without_notification():
|
||||
demo = Demo()
|
||||
make_observable(demo)
|
||||
data = Data()
|
||||
make_observable(data)
|
||||
|
||||
called = []
|
||||
bind(demo, 'number', lambda old, new: called.append(True))
|
||||
bind(data, 'number', lambda old, new: called.append(True))
|
||||
|
||||
demo.other_attr = "value"
|
||||
data.other_attr = "value"
|
||||
|
||||
assert len(called) == 0
|
||||
assert demo.other_attr == "value"
|
||||
assert data.other_attr == "value"
|
||||
|
||||
|
||||
def test_i_can_have_multiple_instances_with_independent_observers():
|
||||
demo1 = Demo()
|
||||
demo2 = Demo()
|
||||
data1 = Data()
|
||||
data2 = Data()
|
||||
|
||||
make_observable(demo1)
|
||||
make_observable(data1)
|
||||
|
||||
calls1 = []
|
||||
bind(demo1, 'number', lambda old, new: calls1.append(new))
|
||||
bind(data1, 'number', lambda old, new: calls1.append(new))
|
||||
|
||||
demo1.number = 5
|
||||
demo2.number = 10
|
||||
data1.number = 5
|
||||
data2.number = 10
|
||||
|
||||
# Only demo1 should trigger callback
|
||||
# Only data1 should trigger callback
|
||||
assert calls1 == [5]
|
||||
assert demo1.number == 5
|
||||
assert demo2.number == 10
|
||||
assert data1.number == 5
|
||||
assert data2.number == 10
|
||||
|
||||
|
||||
def test_i_can_call_make_observable_multiple_times_safely():
|
||||
demo = Demo()
|
||||
data = Data()
|
||||
|
||||
result1 = make_observable(demo)
|
||||
result2 = make_observable(demo)
|
||||
result1 = make_observable(data)
|
||||
result2 = make_observable(data)
|
||||
|
||||
assert result1 is result2
|
||||
assert hasattr(demo, '_listeners')
|
||||
assert hasattr(data, '_listeners')
|
||||
|
||||
# Should still work normally
|
||||
called = []
|
||||
bind(demo, 'number', lambda old, new: called.append(new))
|
||||
demo.number = 5
|
||||
bind(data, 'number', lambda old, new: called.append(new))
|
||||
data.number = 5
|
||||
|
||||
assert len(called) == 1
|
||||
|
||||
|
||||
def test_i_cannot_bind_before_making_observable():
|
||||
demo = Demo()
|
||||
data = Data()
|
||||
|
||||
with pytest.raises(NotObservableError) as exc_info:
|
||||
bind(demo, 'number', lambda old, new: None)
|
||||
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():
|
||||
demo = Demo()
|
||||
make_observable(demo)
|
||||
data = Data()
|
||||
make_observable(data)
|
||||
|
||||
values = []
|
||||
bind(demo, 'new_attr', lambda old, new: values.append((old, new)))
|
||||
bind(data, 'new_attr', lambda old, new: values.append((old, new)))
|
||||
|
||||
demo.new_attr = "first_value"
|
||||
data.new_attr = "first_value"
|
||||
|
||||
assert values == [(None, "first_value")]
|
||||
assert demo.new_attr == "first_value"
|
||||
assert data.new_attr == "first_value"
|
||||
|
||||
|
||||
def test_i_can_preserve_original_class_behavior():
|
||||
demo = Demo()
|
||||
make_observable(demo)
|
||||
data = Data()
|
||||
make_observable(data)
|
||||
|
||||
# Test that original methods still work
|
||||
assert demo.get_double() == 2
|
||||
assert data.get_double() == 2
|
||||
|
||||
demo.number = 5
|
||||
assert demo.get_double() == 10
|
||||
data.number = 5
|
||||
assert data.get_double() == 10
|
||||
|
||||
# Test that isinstance still works
|
||||
assert isinstance(demo, Demo)
|
||||
assert isinstance(data, Data)
|
||||
|
||||
|
||||
def test_i_can_collect_the_updates():
|
||||
demo = Demo()
|
||||
make_observable(demo)
|
||||
data = Data()
|
||||
make_observable(data)
|
||||
|
||||
bind(demo, 'number', lambda old, new: new)
|
||||
bind(demo, 'number', lambda old, new: new * 2)
|
||||
bind(data, 'number', lambda old, new: new)
|
||||
bind(data, 'number', lambda old, new: new * 2)
|
||||
|
||||
demo.number = 5
|
||||
assert collect_return_values(demo) == [5, 10]
|
||||
data.number = 5
|
||||
assert collect_return_values(data) == [5, 10]
|
||||
|
||||
# another time to make sure there is no side effect
|
||||
demo.number = 10
|
||||
assert collect_return_values(demo) == [10, 20]
|
||||
data.number = 10
|
||||
assert collect_return_values(data) == [10, 20]
|
||||
|
||||
|
||||
def test_i_cannot_collect_updates_before_making_observable():
|
||||
demo = Demo()
|
||||
data = Data()
|
||||
|
||||
with pytest.raises(NotObservableError) as exc_info:
|
||||
collect_return_values(demo)
|
||||
collect_return_values(data)
|
||||
|
||||
assert "must be made observable" in str(exc_info.value).lower()
|
||||
|
||||
|
||||
def test_i_can_collect_none():
|
||||
demo = Demo()
|
||||
make_observable(demo)
|
||||
data = Data()
|
||||
make_observable(data)
|
||||
|
||||
bind(demo, 'number', lambda old, new: None)
|
||||
bind(data, 'number', lambda old, new: None)
|
||||
|
||||
demo.number = 5
|
||||
assert collect_return_values(demo) == [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")
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# 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
|
||||
|
||||
|
||||
def test_same_callback_can_be_bound_to_multiple_attributes(observable_data):
|
||||
"""
|
||||
The same callback can be bound to multiple attributes and unbound independently.
|
||||
"""
|
||||
results = []
|
||||
|
||||
def callback(old, new):
|
||||
results.append((old, new))
|
||||
|
||||
bind(observable_data, "value", callback)
|
||||
bind(observable_data, "count", callback)
|
||||
|
||||
observable_data.value = "first"
|
||||
observable_data.count = 1
|
||||
assert len(results) == 2
|
||||
|
||||
unbind(observable_data, "value", callback)
|
||||
|
||||
observable_data.value = "second"
|
||||
observable_data.count = 2
|
||||
assert len(results) == 3 # Only count callback triggered
|
||||
|
||||
Reference in New Issue
Block a user