feat: add support for system event (#264)

* feat: add support for  event

* push to coveralls only once

* force test to fail

* remove failure test

Author: grdsdev

.github/workflows/ci.yml

@@ -34,23 +34,11 @@ jobs:
         with:
           version: latest
 
-      - name: Setup Local Supabase
-        run: |
-          supabase start --workdir tests
-          supabase db reset --workdir tests
-          supabase status --workdir tests -o env > tests/.env \
-            --override-name auth.anon_key=SUPABASE_ANON_KEY \
-            --override-name api.url=SUPABASE_URL
-
-      - name: Wait for Supabase to be ready
-        run: |
-          echo "Waiting for 5 seconds to ensure Supabase is fully initialized..."
-          sleep 5
-
       - name: Run Tests
-        run: make run_tests || make run_tests
+        run: make run_tests
 
       - name: Upload coverage to Coveralls
+        if: ${{ matrix.python-version }} == "3.12"
         uses: coverallsapp/github-action@v2
         with:
           github-token: ${{ secrets.GITHUB_TOKEN }}

Makefile

@@ -12,5 +12,12 @@ tests_pre_commit:
 
 run_tests: tests
 
-tests_only:
+setup_test_infra:
+	supabase start --workdir tests
+	supabase db reset --workdir tests
+	supabase status --workdir tests -o env > tests/.env \
+		--override-name auth.anon_key=SUPABASE_ANON_KEY \
+		--override-name api.url=SUPABASE_URL
+
+tests_only: setup_test_infra
 	poetry run pytest --cov=./ --cov-report=xml --cov-report=html -vv

pyproject.toml

@@ -4,7 +4,7 @@ version = "2.1.0" # {x-release-please-version}
 description = ""
 authors = [
     "Joel Lee <[email protected]>",
-    "Andrew Smith <[email protected]>"
+    "Andrew Smith <[email protected]>",
 ]
 license = "MIT"
 readme = "README.md"
@@ -33,3 +33,7 @@ pytest-cov = "^5.0.0"
 [build-system]
 requires = ["poetry-core>=1.0.0"]
 build-backend = "poetry.core.masonry.api"
+
+[tool.pytest.ini_options]
+asyncio_mode = "strict"
+asyncio_default_fixture_loop_scope = "function"

realtime/_async/channel.py

@@ -388,6 +388,17 @@ def on_postgres_changes(
             callback=lambda payload, _: callback(payload),
         )
 
+    def on_system(
+        self, callback: Callable[[Dict[str, Any], None]]
+    ) -> AsyncRealtimeChannel:
+        """
+        Set up a listener for system events.
+
+        :param callback: The callback function to execute when a system event is received.
+        :return: The Channel instance for method chaining.
+        """
+        return self._on("system", callback=lambda payload, _: callback(payload))
+
     # Presence methods
     async def track(self, user_status: Dict[str, Any]) -> None:
         """

realtime/_async/client.py

@@ -167,7 +167,6 @@ async def _flush_send_buffer(self):
                 await callback()
             self.send_buffer = []
 
-    @ensure_connection
     async def close(self) -> None:
         """
         Close the WebSocket connection.

tests/test_connection.py

@@ -135,13 +135,16 @@ def delete_callback(payload):
         delete_event.set()
 
     subscribed_event = asyncio.Event()
+    system_event = asyncio.Event()
 
     await channel.on_postgres_changes(
         "*", all_changes_callback, table="todos"
     ).on_postgres_changes("INSERT", insert_callback, table="todos").on_postgres_changes(
         "UPDATE", update_callback, table="todos"
     ).on_postgres_changes(
         "DELETE", delete_callback, table="todos"
+    ).on_system(
+        lambda _: system_event.set()
     ).subscribe(
         lambda state, error: (
             subscribed_event.set()
@@ -150,6 +153,8 @@ def delete_callback(payload):
         )
     )
 
+    await asyncio.wait_for(system_event.wait(), 10)
+
     # Wait for the channel to be subscribed
     await asyncio.wait_for(subscribed_event.wait(), 10)