From d62b135cee6196a21b50e23fbc7e62622782d458 Mon Sep 17 00:00:00 2001 From: Kodjo Sossouvi Date: Sun, 2 Nov 2025 18:10:57 +0100 Subject: [PATCH] Added `unbind`, `unbind_all`, `has_listeners` `get_listener_count` --- README.md | 173 +++++++++++- pyproject.toml | 2 +- src/myutils/observable.py | 225 +++++++++++++++ tests/test_observable.py | 566 ++++++++++++++++++++++++++++++++------ 4 files changed, 877 insertions(+), 89 deletions(-) diff --git a/README.md b/README.md index 7d2bed8..9e57dff 100644 --- a/README.md +++ b/README.md @@ -29,14 +29,24 @@ occurs. - `bind(object, attribute, callback)`: Adds a callback function to an observable object's attribute. The callback is triggered when the attribute value changes. +- `unbind(object, attribute, callback)`: + Removes a callback from an observable object's attribute. +- `unbind_all(object, attribute)`: + Removes all callbacks from an observable object's attribute. +- `has_listeners(object, attribute)`: + Checks if an observable object has any listeners for a specific attribute. +- `get_listener_count(object, attribute)`: + Returns the number of listeners for a specific attribute on an observable object. - Exception: `NotObservableError` is raised if you attempt to bind a callback to a non-observable object. #### Example Usage: +#### Basic Usage + Here is an example using the `Observable` module: ```python -from myutils.observable import make_observable, bind +from myutils.observable import make_observable, bind, unbind class Demo: @@ -47,11 +57,162 @@ class Demo: demo = Demo() make_observable(demo) +l = lambda old, new: print(f"Changed from {old} to {new}") + # Bind a callback to 'number' -bind(demo, 'number', lambda old, new: print(f"Changed from {old} to {new}")) +bind(demo, 'number', l) # Updating an attribute triggers the callback demo.number = 42 # Output: Changed from 1 to 42 + +# Unbind the callback +unbind(demo, 'number', l) +demo.number = 43 # No output +``` + +#### Multiple Callbacks + +```python +from dataclasses import dataclass +from myutils.observable import make_observable, bind, unbind + + +@dataclass +class Data: + value: str = "hello" + + +data = Data() + + +def callback1(old, new): + print(f"Callback1: {old} -> {new}") + + +def callback2(old, new): + print(f"Callback2: {old} -> {new}") + + +# Bind both +bind(data, "value", callback1) +bind(data, "value", callback2) + +data.value = "test" +# Prints: +# Callback1: hello -> test +# Callback2: hello -> test + +# Remove only one +unbind(data, "value", callback1) + +data.value = "final" +# Prints: +# Callback2: test -> final +``` + +#### Multiple Callbacks + +```python +from dataclasses import dataclass +from myutils.observable import make_observable, bind, unbind + + +@dataclass +class Data: + value: str = "hello" + + +data = Data() + + +def callback1(old, new): + print(f"Callback1: {old} -> {new}") + + +def callback2(old, new): + print(f"Callback2: {old} -> {new}") + + +# Bind both +bind(data, "value", callback1) +bind(data, "value", callback2) + +data.value = "test" +# Prints: +# Callback1: hello -> test +# Callback2: hello -> test + +# Remove only one +unbind(data, "value", callback1) + +data.value = "final" +# Prints: +# Callback2: test -> final +``` + +#### Using Helper Functions + +```python +from dataclasses import dataclass +from myutils.observable import has_listeners, get_listener_count, unbind_all + + +@dataclass +class Data: + value: str = "hello" + + +data = Data() + +# Check if attribute has listeners +if has_listeners(data, "value"): + print("Value has listeners") + +# Get count +count = get_listener_count(data, "value") +print(f"Value has {count} listeners") + +# Remove all listeners from one attribute +unbind_all(data, "value") + +# Remove all listeners from all attributes +unbind_all(data) +``` + +### Common Pitfalls + +#### ❌ Wrong: Using different callback instance + +```python +bind(data, "value", lambda old, new: print(old)) +unbind(data, "value", lambda old, new: print(old)) # Different lambda! +# The callback is NOT removed because lambdas are different objects +``` + +#### ✅ Correct: Using same callback reference + +```python +callback = lambda old, new: print(old) +bind(data, "value", callback) +unbind(data, "value", callback) # Same reference +# Callback is properly removed +``` + +#### ❌ Wrong: Assuming unbind raises on missing callback + +```python +try: + unbind(data, "value", non_existent_callback) +except Exception: + # This won't execute - unbind fails silently + pass +``` + +#### ✅ Correct: Check before unbinding if needed + +```python +if has_listeners(data, "value"): + unbind(data, "value", callback) ``` --- @@ -110,8 +271,9 @@ MyUtils │ └── __init__.py # Main package initialization ├── tests │ ├── __init__.py # Test package initialization -│ ├── test_observable.py # Tests for Observable module -│ └── test_expando.py # Tests for Expando module +│ ├── test_dummy.py # Tests for Dummy module +│ ├── test_expando.py # Tests for Expando module +│ └── test_observable.py # Tests for Observable module ├── .gitignore # Git ignore file ├── main.py # Application entry point ├── requirements.txt # Project dependencies @@ -156,4 +318,5 @@ Special thanks to the Python and open-source community for their tools, inspirat ## Release History * 0.1.0 : Initial release -* 0.2.0 : Observable results can be collected using `collect_return_values` \ No newline at end of file +* 0.2.0 : Observable results can be collected using `collect_return_values` +* 0.3.0 : Added `unbind`, `unbind_all`, `has_listeners` `get_listener_count` to Observable \ No newline at end of file diff --git a/pyproject.toml b/pyproject.toml index 8e7a2cf..4600384 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "myutils" -version = "0.2.0" +version = "0.3.0" description = "Base useful classes." readme = "README.md" authors = [ diff --git a/src/myutils/observable.py b/src/myutils/observable.py index c3db3c6..5a67e10 100644 --- a/src/myutils/observable.py +++ b/src/myutils/observable.py @@ -46,3 +46,228 @@ def bind(obj, attr_name, callback): if attr_name not in obj._listeners: obj._listeners[attr_name] = [] obj._listeners[attr_name].append(callback) + + +""" +Implementation of unbind() function for myutils.observable module. + +This implementation is adapted to work with the existing make_observable() +and bind() functions that use the _listeners dictionary pattern. +""" + + +def unbind(obj, attr_name, callback): + """ + Remove a specific callback from an observable attribute. + + This function removes a previously registered callback for an attribute + of an observable object. It should be called with the same parameters + that were used with bind(). + + Args: + obj: The observable object (must have been made observable with make_observable()) + attr_name: The attribute name to stop observing + callback: The exact callback function to remove (must be the same object) + + Raises: + NotObservableError: If the object is not observable + + Example: + >>> from myutils.observable import make_observable, bind, unbind + >>> + >>> @dataclass + >>> class Data: + ... value: str = "hello" + >>> + >>> def my_callback(old, new): + ... print(f"Changed from {old} to {new}") + >>> + >>> data = Data() + >>> make_observable(data) + >>> bind(data, "value", my_callback) + >>> + >>> data.value = "world" # Prints: Changed from hello to world + >>> + >>> unbind(data, "value", my_callback) + >>> data.value = "foo" # Does not print anything + + Notes: + - If the callback is not found, this function fails silently + - If the attribute has no callbacks registered, fails silently + - The callback parameter must be the exact same function object + that was passed to bind() + - After removing the last callback for an attribute, the attribute + key remains in _listeners as an empty list (for performance) + """ + if not hasattr(obj, '_listeners'): + raise NotObservableError( + f"Object must be made observable with make_observable() before unbinding" + ) + + # Check if attribute has any listeners + if attr_name not in obj._listeners: + # No listeners for this attribute, nothing to do + return + + # Try to remove the callback + try: + obj._listeners[attr_name].remove(callback) + except ValueError: + # Callback not found in the list, fail silently + pass + + +def unbind_all(obj, attr_name=None): + """ + Remove all callbacks from an object or a specific attribute. + + Args: + obj: The observable object + attr_name: Optional attribute name. If None, removes all callbacks + from all attributes. If specified, removes all callbacks + from that specific attribute only. + + Raises: + NotObservableError: If the object is not observable + + Example: + >>> # Remove all callbacks from a specific attribute + >>> unbind_all(data, "value") + >>> + >>> # Remove all callbacks from all attributes + >>> unbind_all(data) + """ + if not hasattr(obj, '_listeners'): + raise NotObservableError( + f"Object must be made observable with make_observable() before unbinding" + ) + + if attr_name is None: + # Remove all callbacks from all attributes + obj._listeners.clear() + elif attr_name in obj._listeners: + # Remove all callbacks from specific attribute + obj._listeners[attr_name].clear() + + +def has_listeners(obj, attr_name): + """ + Check if an attribute has any listeners. + + Args: + obj: The observable object + attr_name: The attribute name to check + + Returns: + bool: True if the attribute has at least one listener + + Raises: + NotObservableError: If the object is not observable + + Example: + >>> make_observable(data) + >>> bind(data, "value", callback) + >>> has_listeners(data, "value") + True + >>> unbind(data, "value", callback) + >>> has_listeners(data, "value") + False + """ + if not hasattr(obj, '_listeners'): + raise NotObservableError( + f"Object must be made observable with make_observable() before checking listeners" + ) + + return attr_name in obj._listeners and len(obj._listeners[attr_name]) > 0 + + +def get_listener_count(obj, attr_name): + """ + Get the number of listeners for a specific attribute. + + Args: + obj: The observable object + attr_name: The attribute name to check + + Returns: + int: Number of listeners registered for the attribute + + Raises: + NotObservableError: If the object is not observable + + Example: + >>> make_observable(data) + >>> bind(data, "value", callback1) + >>> bind(data, "value", callback2) + >>> get_listener_count(data, "value") + 2 + """ + if not hasattr(obj, '_listeners'): + raise NotObservableError( + f"Object must be made observable with make_observable() before getting listener count" + ) + + if attr_name not in obj._listeners: + return 0 + + return len(obj._listeners[attr_name]) + + +# Complete example of usage +if __name__ == "__main__": + from dataclasses import dataclass + + + @dataclass + class Data: + value: str = "hello" + count: int = 0 + + + # Create observable + data = Data() + make_observable(data) + + + # Define callbacks + def callback1(old, new): + print(f"Callback1: {old} -> {new}") + return "callback1_result" + + + def callback2(old, new): + print(f"Callback2: {old} -> {new}") + return "callback2_result" + + + # Bind callbacks + bind(data, "value", callback1) + bind(data, "value", callback2) + + print("=== Both callbacks active ===") + data.value = "world" + # Prints: + # Callback1: hello -> world + # Callback2: hello -> world + + print(f"Listener count: {get_listener_count(data, 'value')}") # 2 + + # Unbind one callback + unbind(data, "value", callback1) + + print("\n=== Only callback2 active ===") + data.value = "foo" + # Prints: + # Callback2: world -> foo + + print(f"Listener count: {get_listener_count(data, 'value')}") # 1 + + # Unbind remaining callback + unbind(data, "value", callback2) + + print("\n=== No callbacks active ===") + data.value = "bar" + # Prints nothing + + print(f"Has listeners: {has_listeners(data, 'value')}") # False + print(f"Listener count: {get_listener_count(data, 'value')}") # 0 diff --git a/tests/test_observable.py b/tests/test_observable.py index 34636ea..25dd9dd 100644 --- a/tests/test_observable.py +++ b/tests/test_observable.py @@ -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