address review comments and adjust tests

Author: Liam-DeVoe

hypothesis-python/src/hypothesis/database.py

@@ -58,12 +58,12 @@
     from watchdog.observers.api import BaseObserver
 
 StrPathT: "TypeAlias" = Union[str, PathLike[str]]
-ListenerEventType: "TypeAlias" = Literal["save", "delete", "move"]
+ListenerEventType: "TypeAlias" = Literal["save", "delete"]
 SaveDataT: "TypeAlias" = tuple[bytes, bytes]  # key, value
 DeleteDataT: "TypeAlias" = tuple[bytes, Optional[bytes]]  # key, value
 MoveDataT: "TypeAlias" = tuple[bytes, bytes, bytes]  # src, dest, value
 ListenerDataT: "TypeAlias" = Union[SaveDataT, DeleteDataT, MoveDataT]
-ListenerT: "TypeAlias" = Callable[[ListenerEventType, ListenerDataT], Any]
+ListenerT: "TypeAlias" = Callable[[tuple[ListenerEventType, ListenerDataT]], Any]
 
 
 def _usable_dir(path: StrPathT) -> bool:
@@ -143,7 +143,6 @@ class ExampleDatabase(metaclass=_EDMeta):
 
     def __init__(self) -> None:
         self._listeners: list[ListenerT] = []
-        self._listening = False
 
     @abc.abstractmethod
     def save(self, key: bytes, value: bytes) -> None:
@@ -182,8 +181,10 @@ def move(self, src: bytes, dest: bytes, value: bytes) -> None:
 
     def add_listener(self, f: ListenerT, /) -> None:
         """Add a change listener."""
+        had_listeners = bool(self._listeners)
         self._listeners.append(f)
-        self._update_listening()
+        if not had_listeners:
+            self._start_listening()
 
     def remove_listener(self, f: ListenerT, /) -> None:
         """
@@ -193,26 +194,17 @@ def remove_listener(self, f: ListenerT, /) -> None:
         if f not in self._listeners:
             return
         self._listeners.remove(f)
-        self._update_listening()
+        if not self._listeners:
+            self._stop_listening()
 
     def clear_listeners(self) -> None:
         """Remove all change listeners."""
+        had_listeners = bool(self._listeners)
         self._listeners.clear()
-        self._update_listening()
-
-    def _update_listening(self) -> None:
-        # - start listening if we're moving from zero to some listeners
-        # - stop listening if we're moving from some to zero listeners
-        if not self._listening and self._listeners:
-            self._start_listening()
-            self._listening = True
-        elif self._listening and not self._listeners:
+        if had_listeners:
             self._stop_listening()
-            self._listening = False
 
-    def _broadcast_change(
-        self, event_type: ListenerEventType, data: ListenerDataT
-    ) -> None:
+    def _broadcast_change(self, event: tuple[ListenerEventType, ListenerDataT]) -> None:
         """
         Called when a value has been either added to or deleted from a key in
         the underlying database store. event_type is one of "save" or "delete".
@@ -227,7 +219,7 @@ def _broadcast_change(
         for changing the file.
         """
         for listener in self._listeners:
-            listener(event_type, data)
+            listener(event)
 
     def _start_listening(self) -> None:
         """
@@ -285,7 +277,7 @@ def save(self, key: bytes, value: bytes) -> None:
         values.add(value)
 
         if changed:
-            self._broadcast_change("save", (key, value))
+            self._broadcast_change(("save", (key, value)))
 
     def delete(self, key: bytes, value: bytes) -> None:
         value = bytes(value)
@@ -294,7 +286,7 @@ def delete(self, key: bytes, value: bytes) -> None:
         values.discard(value)
 
         if changed:
-            self._broadcast_change("delete", (key, value))
+            self._broadcast_change(("delete", (key, value)))
 
     def _start_listening(self) -> None:
         # declare compatibility with the listener api, but do the actual
@@ -367,7 +359,8 @@ def fetch(self, key: bytes) -> Iterable[bytes]:
     def save(self, key: bytes, value: bytes) -> None:
         key_path = self._key_path(key)
         if key_path.name != self._metakeys_hash:
-            # add this key to our meta entry of all keys, avoiding infinite recursion.
+            # add this key to our meta entry of all keys - taking care to avoid
+            # infinite recursion.
             self.save(self._metakeys_name, key)
 
         # Note: we attempt to create the dir in question now. We
@@ -447,6 +440,7 @@ class Handler(FileSystemEventHandler):
             def on_created(
                 _self, event: Union[FileCreatedEvent, DirCreatedEvent]
             ) -> None:
+                print("create", event)
                 # we only registered for the file creation event
                 assert not isinstance(event, DirCreatedEvent)
                 # watchdog events are only bytes if we passed a byte path to
@@ -472,11 +466,12 @@ def on_created(
                 except OSError:  # pragma: no cover
                     return
 
-                _broadcast_change("save", (key, value))
+                _broadcast_change(("save", (key, value)))
 
             def on_deleted(
                 self, event: Union[FileDeletedEvent, DirDeletedEvent]
             ) -> None:
+                print("delete", event)
                 assert not isinstance(event, DirDeletedEvent)
                 assert isinstance(event.src_path, str)
 
@@ -485,9 +480,10 @@ def on_deleted(
                 if key is None:  # pragma: no cover
                     return
 
-                _broadcast_change("delete", (key, None))
+                _broadcast_change(("delete", (key, None)))
 
             def on_moved(self, event: Union[FileMovedEvent, DirMovedEvent]) -> None:
+                print("move", event)
                 assert not isinstance(event, DirMovedEvent)
                 assert isinstance(event.src_path, str)
                 assert isinstance(event.dest_path, str)
@@ -505,7 +501,8 @@ def on_moved(self, event: Union[FileMovedEvent, DirMovedEvent]) -> None:
                 except OSError:  # pragma: no cover
                     return
 
-                _broadcast_change("move", (k1, k2, value))
+                _broadcast_change(("delete", (k1, value)))
+                _broadcast_change(("save", (k2, value)))
 
         self._observer = Observer()
         self._observer.schedule(

hypothesis-python/src/hypothesis/extra/redis.py

@@ -71,16 +71,19 @@ def _pipeline(
         for key in reset_expire_keys:
             pipe.expire(self._prefix + key, self._expire_after)
         if execute_and_publish:
-            # pipe.execute returns a value for each operation, which includes
-            # whatever we did in the yield as a prefix, and the n operations from
-            # pipe.expire as a suffix. remove that suffix to get just the prefix.
-            values = pipe.execute()
-            values = values[: -len(reset_expire_keys)]
-            if any(value > 0 for value in values):
+            changed = pipe.execute()
+            # pipe.execute returns the rows modified for each operation, which includes
+            # the operations performed during the yield, followed by the n operations
+            # from pipe.exire. Look at just the operations from during the yield.
+            changed = changed[: -len(reset_expire_keys)]
+            if any(count > 0 for count in changed):
                 assert to_publish is not None
                 assert event_type is not None
-                to_publish = (event_type, *(self._encode(v) for v in to_publish))
-                self.redis.publish(self.listener_channel, json.dumps(to_publish))
+                self._publish((event_type, to_publish))
+
+    def _publish(self, event):
+        event = (event[0], tuple(self._encode(v) for v in event[1]))
+        self.redis.publish(self.listener_channel, json.dumps(event))
 
     def _encode(self, value: bytes) -> str:
         return base64.b64encode(value).decode("ascii")
@@ -106,20 +109,24 @@ def move(self, src: bytes, dest: bytes, value: bytes) -> None:
             self.save(dest, value)
             return
 
-        with self._pipeline(
-            src, dest, event_type="move", to_publish=(src, dest, value)
-        ) as pipe:
+        with self._pipeline(src, dest, execute_and_publish=False) as pipe:
             pipe.srem(self._prefix + src, value)
             pipe.sadd(self._prefix + dest, value)
 
+        changed = pipe.execute()
+        if changed[0] > 0:
+            self._publish(("delete", (src, value)))
+        if changed[1] > 0:
+            self._publish(("save", (dest, value)))
+
     def _handle_message(self, message: dict) -> None:
         # other message types include "subscribe" and "unsubscribe". these are
         # sent to the client, but not to the pubsub channel.
         assert message["type"] == "message"
         data = json.loads(message["data"])
         event_type = data[0]
         self._broadcast_change(
-            event_type, tuple(self._decode(v) for v in data[1:])  # type: ignore
+            (event_type, tuple(self._decode(v) for v in data[1]))  # type: ignore
         )
 
     def _start_listening(self) -> None:

hypothesis-python/tests/cover/test_database_backend.py

@@ -515,9 +515,7 @@ def _database_conforms_to_listener_api(
     create_db,
     *,
     flush=None,
-    supports_move=False,
     supports_value_delete=True,
-    save_on_no_k1_in_move=False,
     parent_settings=None,
 ):
     # this function is a big mess to support a bunch of different special cases
@@ -529,13 +527,9 @@ def _database_conforms_to_listener_api(
     # * flush is a callable which takes the instantiated db as an argument, and
     #   is called on every step as an invariant. This lets the database do things
     #   like, time.sleep to give time for events to fire.
-    # * supports_move is True if the db fires a "move" event,
-    #   and False if it fires a "delete" followed by a "save" instead.
     # * suports_value_delete is True if the db supports passing
     #   the exact value of a deleted key in "delete" events. The directory database
     #   notably does not support this, and passes None instead.
-    # * save_on_no_k1_in_move is True if the db fires a "save" rather than "move" if
-    #   the value is not present in the source key.
 
     @settings(parent_settings)
     class TestDatabaseListener(RuleBasedStateMachine):
@@ -553,8 +547,8 @@ def __init__(self):
             self.expected_events = []
             self.actual_events = []
 
-            def listener(event_type, args):
-                self.actual_events.append((event_type, args))
+            def listener(event):
+                self.actual_events.append(event)
 
             self.listener = listener
             self.active_listeners = []
@@ -572,9 +566,6 @@ def _expect_delete(self, k, v):
         def _expect_save(self, k, v):
             self._expect_event("save", (k, v))
 
-        def _expect_move(self, k1, k2, v):
-            self._expect_event("move", (k1, k2, v))
-
         @rule(target=keys, k=st.binary())
         def k(self, k):
             return k
@@ -629,24 +620,17 @@ def move(self, k1, k2, v):
             delete_changed = k1 != k2 and in_k1
             self.db.move(k1, k2, v)
 
-            # in database backends that support it, this gets emitted as a move
-            # (*unless* the keys are equal, in which case it is a save).
-            #
-            # otherwise it gets emitted as a delete followed by a save, unless
-            # those did not change the db.
-            if supports_move:
-                if k1 == k2 and save_changed:
-                    self._expect_save(k2, v)
-                elif save_on_no_k1_in_move and not in_k1 and save_changed:
-                    self._expect_save(k2, v)
-                elif delete_changed or save_changed:
-                    self._expect_move(k1, k2, v)
-            else:
-                if delete_changed:
-                    self._expect_delete(k1, v)
-                if save_changed:
-                    self._expect_save(k2, v)
-
+            # A move gets emitted as a delete followed by a save.  The
+            # delete may be omitted if k1==k2, and the save if v in db.fetch(k2).
+            if delete_changed:
+                self._expect_delete(k1, v)
+            if save_changed:
+                self._expect_save(k2, v)
+
+        # it would be nice if this was an @rule, but that runs into race condition
+        # failures where an event listener is removed immediately after a
+        # save/delete/move operation, before the listener can fire. This is only
+        # relevant for DirectoryBasedExampleDatabase.
         @invariant()
         def events_agree(self):
             if flush is not None:
@@ -673,23 +657,18 @@ def test_database_listener_background_write():
 
 def test_can_remove_nonexistent_listener():
     db = InMemoryExampleDatabase()
-    db.remove_listener(lambda *_: _)
+    db.remove_listener(lambda event: event)
 
 
 class DoesNotSupportListening(ExampleDatabase):
-    def save(self, key: bytes, value: bytes) -> None:
-        pass
-
-    def fetch(self, key: bytes) -> Iterable[bytes]:
-        pass
-
-    def delete(self, key: bytes, value: bytes) -> None:
-        pass
+    def save(self, key: bytes, value: bytes) -> None: ...
+    def fetch(self, key: bytes) -> Iterable[bytes]: ...
+    def delete(self, key: bytes, value: bytes) -> None: ...
 
 
 def test_warns_when_listening_not_supported():
     db = DoesNotSupportListening()
-    listener = lambda *_: _
+    listener = lambda event: event
 
     with pytest.warns(
         HypothesisWarning, match="does not support listening for changes"
@@ -705,7 +684,7 @@ def test_warns_when_listening_not_supported():
 def test_readonly_listener():
     db = ReadOnlyDatabase(InMemoryExampleDatabase())
 
-    def listener(event_type, data):
+    def listener(event):
         raise AssertionError("ReadOnlyDatabase never fires change events")
 
     db.add_listener(listener)
@@ -751,3 +730,45 @@ def test_metakeys(tmp_path):
 
     db.save(b"k2", b"v1")
     assert set(db.fetch(db._metakeys_name)) == {b"k1", b"k2"}
+
+
+class TracksListens(ExampleDatabase):
+    def __init__(self):
+        super().__init__()
+        self.starts = 0
+        self.ends = 0
+
+    def save(self, key: bytes, value: bytes) -> None: ...
+    def fetch(self, key: bytes) -> Iterable[bytes]: ...
+    def delete(self, key: bytes, value: bytes) -> None: ...
+
+    def _start_listening(self):
+        self.starts += 1
+
+    def _stop_listening(self):
+        self.ends += 1
+
+
+def test_start_end_listening():
+    db = TracksListens()
+
+    def listener1(event):
+        pass
+
+    def listener2(event):
+        pass
+
+    assert db.starts == 0
+    db.add_listener(listener1)
+    assert db.starts == 1
+    db.add_listener(listener2)
+    assert db.starts == 1
+
+    assert db.ends == 0
+    db.remove_listener(listener2)
+    assert db.ends == 0
+    db.remove_listener(listener1)
+    assert db.ends == 1
+
+    db.clear_listeners()
+    assert db.ends == 1

hypothesis-python/tests/redis/test_redis_exampledatabase.py

@@ -109,14 +109,13 @@ def test_redis_listener():
     _database_conforms_to_listener_api(
         lambda _path: RedisExampleDatabase(FakeRedis()),
         flush=flush_messages,
-        supports_move=True,
     )
 
 
 def test_redis_listener_explicit():
     calls = 0
 
-    def listener(event_type, data):
+    def listener(event):
         nonlocal calls
         calls += 1