fix: fix parent_conn not initialize

.github/workflows/release.yml

@@ -11,8 +11,8 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        python-version: [ 3.5, 3.6, 3.7, 3.8, 3.9 ]
-        os: [ ubuntu-latest, macOS-latest ]
+        python-version: [ 3.6, 3.7, 3.8, 3.9 ]
+        os: [ ubuntu-latest, macOS-latest, windows-latest]
 
     steps:
       - name: Checkout
@@ -23,12 +23,40 @@ jobs:
         with:
           python-version: ${{ matrix.python-version }}
 
-      - name: Setup PostgreSQL
-        uses: Harmon758/postgresql-action@v1.0.0
-        with:
-          postgresql db: postgres
-          postgresql user: postgres
-          postgresql password: 123456
+      - name: Set up PostgreSQL on Linux
+        if: startsWith(runner.os, 'linux')
+        run: |
+          sudo systemctl start postgresql.service
+          pg_isready
+          sudo -u postgres psql --command="ALTER USER postgres WITH PASSWORD '123456'" --command="\du"
+
+      - name: Setup PostgreSQL on macOS
+        if: startsWith(runner.os, 'macos')
+        run: |
+          brew services start postgresql
+          echo "Check PostgreSQL service is running"
+          i=10
+          COMMAND='pg_isready'
+          while [ $i -gt 0 ]; do
+              echo "Check PostgreSQL service status"
+              eval $COMMAND && break
+              ((i--))
+              if [ $i == 0 ]; then
+                  echo "PostgreSQL service not ready, all attempts exhausted"
+                  exit 1
+              fi
+              echo "PostgreSQL service not ready, wait 10 more sec, attempts left: $i"
+              sleep 10
+          done
+          psql --command="CREATE USER postgres PASSWORD '123456'" --command="\du" postgres
+
+      - name: Start PostgreSQL on Windows
+        if: startsWith(runner.os, 'windows')
+        run: |
+          $pgService = Get-Service -Name postgresql*
+          Set-Service -InputObject $pgService -Status running -StartupType automatic
+          Start-Process -FilePath "$env:PGBIN\pg_isready" -Wait -PassThru
+          & $env:PGBIN\psql --command="ALTER USER postgres WITH PASSWORD '123456'" --command="\du"
 
       - name: Install dependencies
         run: |
@@ -54,13 +82,15 @@ jobs:
         run: |
           pip install -r requirements.txt
           pip install -r dev_requirements.txt
+          pip install coveralls
+          pip install coverage
 
-      - name: Setup PostgreSQL
-        uses: Harmon758/postgresql-action@v1.0.0
-        with:
-          postgresql db: postgres
-          postgresql user: postgres
-          postgresql password: 123456
+      - name: Set up PostgreSQL on Linux
+        if: startsWith(runner.os, 'linux')
+        run: |
+          sudo systemctl start postgresql.service
+          pg_isready
+          sudo -u postgres psql --command="ALTER USER postgres WITH PASSWORD '123456'" --command="\du"
 
       - name: Run tests
         run: coverage run -m unittest discover -s tests -t tests

postgresql_watcher/watcher.py

@@ -8,7 +8,7 @@
 
 
 def casbin_subscription(
-    process_conn: connection.PipeConnection,
+    process_conn: Pipe,
     host: str,
     user: str,
     password: str,
@@ -34,7 +34,6 @@ def casbin_subscription(
                 process_conn.put(notify.payload)
 
 
-
 class PostgresqlWatcher(object):
     def __init__(
         self,
@@ -45,21 +44,23 @@ def __init__(
         channel_name: Optional[str] = POSTGRESQL_CHANNEL_NAME,
         start_process: Optional[bool] = True,
     ):
+        self.update_callback = None
+        self.parent_conn = None
         self.host = host
         self.port = port
         self.user = user
         self.password = password
         self.channel_name = channel_name
-        self.subscribed_process, self.parent_conn = self.create_subscriber_process(
-            start_process
-        )
+        self.subscribed_process = self.create_subscriber_process(start_process)
 
     def create_subscriber_process(
         self,
         start_process: Optional[bool] = True,
         delay: Optional[int] = 2,
     ):
         parent_conn, child_conn = Pipe()
+        if not self.parent_conn:
+            self.parent_conn = parent_conn
         p = Process(
             target=casbin_subscription,
             args=(
@@ -76,16 +77,12 @@ def create_subscriber_process(
         if start_process:
             p.start()
         self.should_reload()
-        return p, parent_conn
+        return p
 
-    def update_callback(self):
-        print("callback called because casbin role updated")
-
-    def set_update_callback(self, fn_name: Any):
-        print("runtime is set update callback",fn_name)
+    def set_update_callback(self, fn_name: Callable):
+        print("runtime is set update callback", fn_name)
         self.update_callback = fn_name
 
-
     def update(self):
         conn = connect(
             host=self.host,
@@ -116,4 +113,4 @@ def should_reload(self):
             self.subscribed_process, self.parent_conn = self.create_subscriber_process(
                 delay=10
             )
-            return False
+            return False

tests/test_postgresql_watcher.py

@@ -1,4 +1,7 @@
+import sys
 import unittest
+from multiprocessing.connection import Pipe
+
 from postgresql_watcher import PostgresqlWatcher
 from multiprocessing import connection, context
 
@@ -10,16 +13,26 @@
 
 
 def get_watcher():
-    pg_watcher = PostgresqlWatcher(host=HOST, port=PORT, user=USER, password=PASSWORD)
-    return pg_watcher
+    return PostgresqlWatcher(host=HOST, port=PORT, user=USER, password=PASSWORD)
 
 
 pg_watcher = get_watcher()
 
+try:
+    import _winapi
+    from _winapi import WAIT_OBJECT_0, WAIT_ABANDONED_0, WAIT_TIMEOUT, INFINITE
+except ImportError:
+    if sys.platform == 'win32':
+        raise
+    _winapi = None
+
 
 class TestConfig(unittest.TestCase):
     def test_pg_watcher_init(self):
-        assert isinstance(pg_watcher.parent_conn, connection.PipeConnection)
+        if _winapi:
+            assert isinstance(pg_watcher.parent_conn, connection.PipeConnection)
+        else:
+            assert isinstance(pg_watcher.parent_conn, connection.Connection)
         assert isinstance(pg_watcher.subscribed_process, context.Process)
 
     def test_update_pg_watcher(self):
@@ -40,4 +53,4 @@ def _test_callback():
 
 
 if __name__ == "__main__":
-    unittest.main()
+    unittest.main()