From 1008ca974430eb5ccb72a323c101377ce207e58a Mon Sep 17 00:00:00 2001 From: Taus Date: Mon, 25 Aug 2025 14:14:16 +0000 Subject: [PATCH] Python: Add `psycopg2.pool` tests --- .../frameworks/psycopg2/ConceptsTest.expected | 0 .../frameworks/psycopg2/ConceptsTest.ql | 2 + .../frameworks/psycopg2/connectionpool.py | 46 +++++++++++++++++++ 3 files changed, 48 insertions(+) create mode 100644 python/ql/test/library-tests/frameworks/psycopg2/ConceptsTest.expected create mode 100644 python/ql/test/library-tests/frameworks/psycopg2/ConceptsTest.ql create mode 100644 python/ql/test/library-tests/frameworks/psycopg2/connectionpool.py diff --git a/python/ql/test/library-tests/frameworks/psycopg2/ConceptsTest.expected b/python/ql/test/library-tests/frameworks/psycopg2/ConceptsTest.expected new file mode 100644 index 00000000000..e69de29bb2d diff --git a/python/ql/test/library-tests/frameworks/psycopg2/ConceptsTest.ql b/python/ql/test/library-tests/frameworks/psycopg2/ConceptsTest.ql new file mode 100644 index 00000000000..b557a0bccb6 --- /dev/null +++ b/python/ql/test/library-tests/frameworks/psycopg2/ConceptsTest.ql @@ -0,0 +1,2 @@ +import python +import experimental.meta.ConceptsTest diff --git a/python/ql/test/library-tests/frameworks/psycopg2/connectionpool.py b/python/ql/test/library-tests/frameworks/psycopg2/connectionpool.py new file mode 100644 index 00000000000..507cdd59b82 --- /dev/null +++ b/python/ql/test/library-tests/frameworks/psycopg2/connectionpool.py @@ -0,0 +1,46 @@ +# Examples using psycopg2 connection pools. + +import psycopg2 +from psycopg2.pool import SimpleConnectionPool, AbstractConnectionPool + + +DSN = "dbname=test user=test password=test host=localhost port=5432" + + +def run_simple_pool_query(): + pool = SimpleConnectionPool(1, 4, dsn=DSN) + try: + conn = pool.getconn() + try: + cur = conn.cursor() + try: + # Simple, parameterless query + cur.execute("SELECT 1") # $ getSql="SELECT 1" + _ = cur.fetchall() if hasattr(cur, "fetchall") else None # $ threatModelSource[database]=cur.fetchall() + finally: + cur.close() + finally: + pool.putconn(conn) + finally: + pool.closeall() + + +class LocalPool(AbstractConnectionPool): + pass + + +def run_custom_pool_query(): + pool = LocalPool(1, 3, dsn=DSN) + try: + conn = pool.getconn() + try: + cur = conn.cursor() + try: + cur.execute("SELECT 2") # $ getSql="SELECT 2" + _ = cur.fetchone() if hasattr(cur, "fetchone") else None # $ threatModelSource[database]=cur.fetchone() + finally: + cur.close() + finally: + pool.putconn(conn) + finally: + pool.closeall()