chore: Add tests for Reference.listen() (firebase#851)

jonathanedey · web-flow · commit e6c95e7ef6e8 · 2025-01-22T12:06:20.000-05:00
* chore: Add unit tests for `Reference.listen()`

* Integration test for rtdb listeners

* fix lint
diff --git a/integration/test_db.py b/integration/test_db.py
@@ -16,6 +16,7 @@
 import collections
 import json
 import os
+import time
 
 import pytest
 
@@ -245,6 +246,37 @@ def test_delete(self, testref):
         ref.delete()
         assert ref.get() is None
 
+class TestListenOperations:
+    """Test cases for listening to changes to node values."""
+
+    def test_listen(self, testref):
+        self.events = []
+        def callback(event):
+            self.events.append(event)
+
+        python = testref.parent
+        registration = python.listen(callback)
+        try:
+            ref = python.child('users').push()
+            assert ref.path == '/_adminsdk/python/users/' + ref.key
+            assert ref.get() == ''
+
+            self.wait_for(self.events, count=2)
+            assert len(self.events) == 2
+
+            assert self.events[1].event_type == 'put'
+            assert self.events[1].path == '/users/' + ref.key
+            assert self.events[1].data == ''
+        finally:
+            registration.close()
+
+    @classmethod
+    def wait_for(cls, events, count=1, timeout_seconds=5):
+        must_end = time.time() + timeout_seconds
+        while time.time() < must_end:
+            if len(events) >= count:
+                return
+        raise pytest.fail('Timed out while waiting for events')
 
 class TestAdvancedQueries:
     """Test cases for advanced interactions via the db.Query interface."""
diff --git a/tests/test_db.py b/tests/test_db.py
@@ -535,6 +535,49 @@ def callback(_):
         finally:
             testutils.cleanup_apps()
 
+    @pytest.mark.parametrize(
+        'url,emulator_host,expected_base_url,expected_namespace',
+        [
+            # Production URLs with no override:
+            ('https://test.firebaseio.com', None, 'https://test.firebaseio.com/.json', None),
+            ('https://test.firebaseio.com/', None, 'https://test.firebaseio.com/.json', None),
+
+            # Production URLs with emulator_host override:
+            ('https://test.firebaseio.com', 'localhost:9000', 'http://localhost:9000/.json',
+             'test'),
+            ('https://test.firebaseio.com/', 'localhost:9000', 'http://localhost:9000/.json',
+             'test'),
+
+            # Emulator URL with no override.
+            ('http://localhost:8000/?ns=test', None, 'http://localhost:8000/.json', 'test'),
+
+            # emulator_host is ignored when the original URL is already emulator.
+            ('http://localhost:8000/?ns=test', 'localhost:9999', 'http://localhost:8000/.json',
+             'test'),
+        ]
+    )
+    def test_listen_sse_client(self, url, emulator_host, expected_base_url, expected_namespace,
+                               mocker):
+        if emulator_host:
+            os.environ[_EMULATOR_HOST_ENV_VAR] = emulator_host
+
+        try:
+            firebase_admin.initialize_app(testutils.MockCredential(), {'databaseURL' : url})
+            ref = db.reference()
+            mock_sse_client = mocker.patch('firebase_admin._sseclient.SSEClient')
+            mock_callback = mocker.Mock()
+            ref.listen(mock_callback)
+            args, kwargs = mock_sse_client.call_args
+            assert args[0] == expected_base_url
+            if expected_namespace:
+                assert kwargs.get('params') == {'ns': expected_namespace}
+            else:
+                assert kwargs.get('params') == {}
+        finally:
+            if _EMULATOR_HOST_ENV_VAR in os.environ:
+                del os.environ[_EMULATOR_HOST_ENV_VAR]
+            testutils.cleanup_apps()
+
     def test_listener_session(self):
         firebase_admin.initialize_app(testutils.MockCredential(), {
             'databaseURL' : 'https://test.firebaseio.com',
