link gif

.circleci/run_tests.sh

@@ -106,7 +106,7 @@ cd ../..
 # These tests will start and stop the pgcat server so it will need to be restarted after the tests
 #
 pip3 install -r tests/python/requirements.txt
-pytest || exit 1
+python3 tests/python/tests.py || exit 1
 
 
 #

.gitignore

@@ -11,4 +11,3 @@ dev/.bash_history
 dev/cache
 !dev/cache/.keepme
 .venv
-**/__pycache__

CONTRIBUTING.md

@@ -21,7 +21,7 @@ Within this test environment you can modify the file in your favorite IDE and re
 
 Once the environment is ready, you can run the tests by running
 Ruby:   `cd /app/tests/ruby && bundle exec ruby <test_name>.rb --format documentation`
-Python: `cd /app/ && pytest`
+Python: `cd /app && python3 tests/python/tests.py`
 Rust:   `cd /app/tests/rust && cargo run`
 Go:     `cd /app/tests/go && /usr/local/go/bin/go test`
 

start_test_env.sh

@@ -23,7 +23,7 @@ docker compose exec --workdir /app/tests/python main pip3 install -r requirement
 echo "Interactive test environment ready"
 echo "To run integration tests, you can use the following commands:"
 echo -e "   ${BLUE}Ruby:   ${RED}cd /app/tests/ruby && bundle exec ruby tests.rb --format documentation${RESET}"
-echo -e "   ${BLUE}Python: ${RED}cd /app/ && pytest ${RESET}"
+echo -e "   ${BLUE}Python: ${RED}cd /app && python3 tests/python/tests.py${RESET}"
 echo -e "   ${BLUE}Rust:   ${RED}cd /app/tests/rust && cargo run ${RESET}"
 echo -e "   ${BLUE}Go:     ${RED}cd /app/tests/go && /usr/local/go/bin/go test${RESET}"
 echo "the source code for tests are directly linked to the source code in the container so you can modify the code and run the tests again"

tests/python/requirements.txt

@@ -1,3 +1,2 @@
-pytest
 psycopg2==2.9.3
 psutil==5.9.1

tests/python/tests.py

@@ -1,29 +1,83 @@
+from typing import Tuple
+import psycopg2
+import psutil
 import os
 import signal
 import time
 
-import psycopg2
-
-import utils
-
 SHUTDOWN_TIMEOUT = 5
 
+PGCAT_HOST = "127.0.0.1"
+PGCAT_PORT = "6432"
+
+
+def pgcat_start():
+    pg_cat_send_signal(signal.SIGTERM)
+    os.system("./target/debug/pgcat .circleci/pgcat.toml &")
+    time.sleep(2)
+
+
+def pg_cat_send_signal(signal: signal.Signals):
+    try:
+        for proc in psutil.process_iter(["pid", "name"]):
+            if "pgcat" == proc.name():
+                os.kill(proc.pid, signal)
+    except Exception as e:
+        # The process can be gone when we send this signal
+        print(e)
+
+    if signal == signal.SIGTERM:
+        # Returns 0 if pgcat process exists
+        time.sleep(2)
+        if not os.system('pgrep pgcat'):
+            raise Exception("pgcat not closed after SIGTERM")
+
+
+def connect_db(
+    autocommit: bool = True,
+    admin: bool = False,
+) -> Tuple[psycopg2.extensions.connection, psycopg2.extensions.cursor]:
+
+    if admin:
+        user = "admin_user"
+        password = "admin_pass"
+        db = "pgcat"
+    else:
+        user = "sharding_user"
+        password = "sharding_user"
+        db = "sharded_db"
+
+    conn = psycopg2.connect(
+        f"postgres://{user}:{password}@{PGCAT_HOST}:{PGCAT_PORT}/{db}?application_name=testing_pgcat",
+        connect_timeout=2,
+    )
+    conn.autocommit = autocommit
+    cur = conn.cursor()
+
+    return (conn, cur)
+
+
+def cleanup_conn(conn: psycopg2.extensions.connection, cur: psycopg2.extensions.cursor):
+    cur.close()
+    conn.close()
+
+
 def test_normal_db_access():
-    utils.pgcat_start()
-    conn, cur = utils.connect_db(autocommit=False)
+    pgcat_start()
+    conn, cur = connect_db(autocommit=False)
     cur.execute("SELECT 1")
     res = cur.fetchall()
     print(res)
-    utils.cleanup_conn(conn, cur)
+    cleanup_conn(conn, cur)
 
 
 def test_admin_db_access():
-    conn, cur = utils.connect_db(admin=True)
+    conn, cur = connect_db(admin=True)
 
     cur.execute("SHOW POOLS")
     res = cur.fetchall()
     print(res)
-    utils.cleanup_conn(conn, cur)
+    cleanup_conn(conn, cur)
 
 
 def test_shutdown_logic():
@@ -32,17 +86,17 @@ def test_shutdown_logic():
     # NO ACTIVE QUERIES SIGINT HANDLING
 
     # Start pgcat
-    utils.pgcat_start()
+    pgcat_start()
 
     # Create client connection and send query (not in transaction)
-    conn, cur = utils.connect_db()
+    conn, cur = connect_db()
 
     cur.execute("BEGIN;")
     cur.execute("SELECT 1;")
     cur.execute("COMMIT;")
 
     # Send sigint to pgcat
-    utils.pg_cat_send_signal(signal.SIGINT)
+    pg_cat_send_signal(signal.SIGINT)
     time.sleep(1)
 
     # Check that any new queries fail after sigint since server should close with no active transactions
@@ -54,18 +108,18 @@ def test_shutdown_logic():
         # Fail if query execution succeeded
         raise Exception("Server not closed after sigint")
 
-    utils.cleanup_conn(conn, cur)
-    utils.pg_cat_send_signal(signal.SIGTERM)
+    cleanup_conn(conn, cur)
+    pg_cat_send_signal(signal.SIGTERM)
 
     # - - - - - - - - - - - - - - - - - -
     # NO ACTIVE QUERIES ADMIN SHUTDOWN COMMAND
 
     # Start pgcat
-    utils.pgcat_start()
+    pgcat_start()
 
     # Create client connection and begin transaction
-    conn, cur = utils.connect_db()
-    admin_conn, admin_cur = utils.connect_db(admin=True)
+    conn, cur = connect_db()
+    admin_conn, admin_cur = connect_db(admin=True)
 
     cur.execute("BEGIN;")
     cur.execute("SELECT 1;")
@@ -84,24 +138,24 @@ def test_shutdown_logic():
         # Fail if query execution succeeded
         raise Exception("Server not closed after sigint")
 
-    utils.cleanup_conn(conn, cur)
-    utils.cleanup_conn(admin_conn, admin_cur)
-    utils.pg_cat_send_signal(signal.SIGTERM)
+    cleanup_conn(conn, cur)
+    cleanup_conn(admin_conn, admin_cur)
+    pg_cat_send_signal(signal.SIGTERM)
 
     # - - - - - - - - - - - - - - - - - -
     # HANDLE TRANSACTION WITH SIGINT
 
     # Start pgcat
-    utils.pgcat_start()
+    pgcat_start()
 
     # Create client connection and begin transaction
-    conn, cur = utils.connect_db()
+    conn, cur = connect_db()
 
     cur.execute("BEGIN;")
     cur.execute("SELECT 1;")
 
     # Send sigint to pgcat while still in transaction
-    utils.pg_cat_send_signal(signal.SIGINT)
+    pg_cat_send_signal(signal.SIGINT)
     time.sleep(1)
 
     # Check that any new queries succeed after sigint since server should still allow transaction to complete
@@ -111,18 +165,18 @@ def test_shutdown_logic():
         # Fail if query fails since server closed
         raise Exception("Server closed while in transaction", e.pgerror)
 
-    utils.cleanup_conn(conn, cur)
-    utils.pg_cat_send_signal(signal.SIGTERM)
+    cleanup_conn(conn, cur)
+    pg_cat_send_signal(signal.SIGTERM)
 
     # - - - - - - - - - - - - - - - - - -
     # HANDLE TRANSACTION WITH ADMIN SHUTDOWN COMMAND
 
     # Start pgcat
-    utils.pgcat_start()
+    pgcat_start()
 
     # Create client connection and begin transaction
-    conn, cur = utils.connect_db()
-    admin_conn, admin_cur = utils.connect_db(admin=True)
+    conn, cur = connect_db()
+    admin_conn, admin_cur = connect_db(admin=True)
 
     cur.execute("BEGIN;")
     cur.execute("SELECT 1;")
@@ -140,30 +194,30 @@ def test_shutdown_logic():
         # Fail if query fails since server closed
         raise Exception("Server closed while in transaction", e.pgerror)
 
-    utils.cleanup_conn(conn, cur)
-    utils.cleanup_conn(admin_conn, admin_cur)
-    utils.pg_cat_send_signal(signal.SIGTERM)
+    cleanup_conn(conn, cur)
+    cleanup_conn(admin_conn, admin_cur)
+    pg_cat_send_signal(signal.SIGTERM)
 
     # - - - - - - - - - - - - - - - - - -
     # NO NEW NON-ADMIN CONNECTIONS DURING SHUTDOWN
     # Start pgcat
-    utils.pgcat_start()
+    pgcat_start()
 
     # Create client connection and begin transaction
-    transaction_conn, transaction_cur = utils.connect_db()
+    transaction_conn, transaction_cur = connect_db()
 
     transaction_cur.execute("BEGIN;")
     transaction_cur.execute("SELECT 1;")
 
     # Send sigint to pgcat while still in transaction
-    utils.pg_cat_send_signal(signal.SIGINT)
+    pg_cat_send_signal(signal.SIGINT)
     time.sleep(1)
 
     start = time.perf_counter()
     try:
-        conn, cur = utils.connect_db()
+        conn, cur = connect_db()
         cur.execute("SELECT 1;")
-        utils.cleanup_conn(conn, cur)
+        cleanup_conn(conn, cur)
     except psycopg2.OperationalError as e:
         time_taken = time.perf_counter() - start
         if time_taken > 0.1:
@@ -173,49 +227,49 @@ def test_shutdown_logic():
     else:
         raise Exception("Able connect to database during shutdown")
 
-    utils.cleanup_conn(transaction_conn, transaction_cur)
-    utils.pg_cat_send_signal(signal.SIGTERM)
+    cleanup_conn(transaction_conn, transaction_cur)
+    pg_cat_send_signal(signal.SIGTERM)
 
     # - - - - - - - - - - - - - - - - - -
     # ALLOW NEW ADMIN CONNECTIONS DURING SHUTDOWN
     # Start pgcat
-    utils.pgcat_start()
+    pgcat_start()
 
     # Create client connection and begin transaction
-    transaction_conn, transaction_cur = utils.connect_db()
+    transaction_conn, transaction_cur = connect_db()
 
     transaction_cur.execute("BEGIN;")
     transaction_cur.execute("SELECT 1;")
 
     # Send sigint to pgcat while still in transaction
-    utils.pg_cat_send_signal(signal.SIGINT)
+    pg_cat_send_signal(signal.SIGINT)
     time.sleep(1)
 
     try:
-        conn, cur = utils.connect_db(admin=True)
+        conn, cur = connect_db(admin=True)
         cur.execute("SHOW DATABASES;")
-        utils.cleanup_conn(conn, cur)
+        cleanup_conn(conn, cur)
     except psycopg2.OperationalError as e:
         raise Exception(e)
 
-    utils.cleanup_conn(transaction_conn, transaction_cur)
-    utils.pg_cat_send_signal(signal.SIGTERM)
+    cleanup_conn(transaction_conn, transaction_cur)
+    pg_cat_send_signal(signal.SIGTERM)
 
     # - - - - - - - - - - - - - - - - - -
     # ADMIN CONNECTIONS CONTINUING TO WORK AFTER SHUTDOWN
     # Start pgcat
-    utils.pgcat_start()
+    pgcat_start()
 
     # Create client connection and begin transaction
-    transaction_conn, transaction_cur = utils.connect_db()
+    transaction_conn, transaction_cur = connect_db()
     transaction_cur.execute("BEGIN;")
     transaction_cur.execute("SELECT 1;")
 
-    admin_conn, admin_cur = utils.connect_db(admin=True)
+    admin_conn, admin_cur = connect_db(admin=True)
     admin_cur.execute("SHOW DATABASES;")
 
     # Send sigint to pgcat while still in transaction
-    utils.pg_cat_send_signal(signal.SIGINT)
+    pg_cat_send_signal(signal.SIGINT)
     time.sleep(1)
 
     try:
@@ -223,24 +277,24 @@ def test_shutdown_logic():
     except psycopg2.OperationalError as e:
         raise Exception("Could not execute admin command:", e)
 
-    utils.cleanup_conn(transaction_conn, transaction_cur)
-    utils.cleanup_conn(admin_conn, admin_cur)
-    utils.pg_cat_send_signal(signal.SIGTERM)
+    cleanup_conn(transaction_conn, transaction_cur)
+    cleanup_conn(admin_conn, admin_cur)
+    pg_cat_send_signal(signal.SIGTERM)
 
     # - - - - - - - - - - - - - - - - - -
     # HANDLE SHUTDOWN TIMEOUT WITH SIGINT
 
     # Start pgcat
-    utils.pgcat_start()
+    pgcat_start()
 
     # Create client connection and begin transaction, which should prevent server shutdown unless shutdown timeout is reached
-    conn, cur = utils.connect_db()
+    conn, cur = connect_db()
 
     cur.execute("BEGIN;")
     cur.execute("SELECT 1;")
 
     # Send sigint to pgcat while still in transaction
-    utils.pg_cat_send_signal(signal.SIGINT)
+    pg_cat_send_signal(signal.SIGINT)
 
     # pgcat shutdown timeout is set to SHUTDOWN_TIMEOUT seconds, so we sleep for SHUTDOWN_TIMEOUT + 1 seconds
     time.sleep(SHUTDOWN_TIMEOUT + 1)
@@ -254,5 +308,12 @@ def test_shutdown_logic():
         # Fail if query execution succeeded
         raise Exception("Server not closed after sigint and expected timeout")
 
-    utils.cleanup_conn(conn, cur)
-    utils.pg_cat_send_signal(signal.SIGTERM)
+    cleanup_conn(conn, cur)
+    pg_cat_send_signal(signal.SIGTERM)
+
+    # - - - - - - - - - - - - - - - - - -
+
+
+test_normal_db_access()
+test_admin_db_access()
+test_shutdown_logic()

tests/python/utils.py

@@ -1,60 +0,0 @@
-from typing import Tuple
-import os
-import psutil
-import signal
-import time
-
-import psycopg2
-
-PGCAT_HOST = "127.0.0.1"
-PGCAT_PORT = "6432"
-
-def pgcat_start():
-    pg_cat_send_signal(signal.SIGTERM)
-    os.system("./target/debug/pgcat .circleci/pgcat.toml &")
-    time.sleep(2)
-
-
-def pg_cat_send_signal(signal: signal.Signals):
-    try:
-        for proc in psutil.process_iter(["pid", "name"]):
-            if "pgcat" == proc.name():
-                os.kill(proc.pid, signal)
-    except Exception as e:
-        # The process can be gone when we send this signal
-        print(e)
-
-    if signal == signal.SIGTERM:
-        # Returns 0 if pgcat process exists
-        time.sleep(2)
-        if not os.system('pgrep pgcat'):
-            raise Exception("pgcat not closed after SIGTERM")
-
-
-def connect_db(
-    autocommit: bool = True,
-    admin: bool = False,
-) -> Tuple[psycopg2.extensions.connection, psycopg2.extensions.cursor]:
-
-    if admin:
-        user = "admin_user"
-        password = "admin_pass"
-        db = "pgcat"
-    else:
-        user = "sharding_user"
-        password = "sharding_user"
-        db = "sharded_db"
-
-    conn = psycopg2.connect(
-        f"postgres://{user}:{password}@{PGCAT_HOST}:{PGCAT_PORT}/{db}?application_name=testing_pgcat",
-        connect_timeout=2,
-    )
-    conn.autocommit = autocommit
-    cur = conn.cursor()
-
-    return (conn, cur)
-
-
-def cleanup_conn(conn: psycopg2.extensions.connection, cur: psycopg2.extensions.cursor):
-    cur.close()
-    conn.close()