Improved tests to require explicit db config and share connection fixtures.

Author: pantierra

.github/workflows/tests/test_notifications.py

@@ -10,6 +10,33 @@
 from datetime import datetime
 
 
+@pytest.fixture(scope='session')
+def db_connection():
+    """Create database connection for testing."""
+    # Require all database connection parameters to be explicitly set
+    required_vars = ['PGHOST', 'PGPORT', 'PGDATABASE', 'PGUSER', 'PGPASSWORD']
+    missing_vars = [var for var in required_vars if not os.getenv(var)]
+
+    if missing_vars:
+        pytest.fail(f"Required environment variables not set: {', '.join(missing_vars)}")
+
+    connection_params = {
+        'host': os.getenv('PGHOST'),
+        'port': int(os.getenv('PGPORT')),
+        'database': os.getenv('PGDATABASE'),
+        'user': os.getenv('PGUSER'),
+        'password': os.getenv('PGPASSWORD')
+    }
+
+    try:
+        conn = psycopg2.connect(**connection_params)
+        conn.set_isolation_level(psycopg2.extensions.ISOLATION_LEVEL_AUTOCOMMIT)
+        yield conn
+        conn.close()
+    except psycopg2.Error as e:
+        pytest.fail(f"Cannot connect to database: {e}")
+
+
 def test_eoapi_notifier_deployment():
     """Test that eoapi-notifier deployment is running."""
     # Check if eoapi-notifier deployment exists and is ready
@@ -98,23 +125,9 @@ def test_eoapi_notifier_logs_show_connection():
     assert "Authentication failed" not in logs, "Should not have auth errors"
 
 
-def test_database_notification_triggers_exist():
+def test_database_notification_triggers_exist(db_connection):
     """Test that pgstac notification triggers are installed."""
-    # Port forward to database if not already done
-    import psycopg2
-    import os
-
-    try:
-        # Try to connect using environment variables set by the test runner
-        conn = psycopg2.connect(
-            host=os.getenv('PGHOST', 'localhost'),
-            port=int(os.getenv('PGPORT', '5432')),
-            database=os.getenv('PGDATABASE', 'eoapi'),
-            user=os.getenv('PGUSER', 'eoapi'),
-            password=os.getenv('PGPASSWORD', '')
-        )
-
-        with conn.cursor() as cur:
+    with db_connection.cursor() as cur:
             # Check if the notification function exists
             cur.execute("""
                 SELECT EXISTS(
@@ -139,13 +152,10 @@ def test_database_notification_triggers_exist():
             trigger_count = result[0] if result else 0
             assert trigger_count >= 3, f"Should have at least 3 triggers (INSERT, UPDATE, DELETE), found {trigger_count}"
 
-        conn.close()
 
-    except (psycopg2.Error, ConnectionError, OSError):
-        pytest.skip("Cannot connect to database for trigger verification")
 
 
-def test_end_to_end_notification_flow():
+def test_end_to_end_notification_flow(db_connection):
     """Test complete flow: database → eoapi-notifier → Knative CloudEvents sink."""
 
     # Skip if notifications not enabled
@@ -160,23 +170,10 @@ def test_end_to_end_notification_flow():
 
     sink_pod = result.stdout.strip()
 
-    # Connect to database using test environment
-    try:
-        conn = psycopg2.connect(
-            host=os.getenv('PGHOST', 'localhost'),
-            port=int(os.getenv('PGPORT', '5432')),
-            database=os.getenv('PGDATABASE', 'eoapi'),
-            user=os.getenv('PGUSER', 'eoapi'),
-            password=os.getenv('PGPASSWORD', '')
-        )
-        conn.set_isolation_level(psycopg2.extensions.ISOLATION_LEVEL_AUTOCOMMIT)
-    except psycopg2.Error as e:
-        pytest.skip(f"Cannot connect to database: {e}")
-
     # Insert test item and check for CloudEvent
     test_item_id = f"e2e-test-{int(time.time())}"
     try:
-        with conn.cursor() as cursor:
+        with db_connection.cursor() as cursor:
             cursor.execute("SELECT pgstac.create_item(%s);", (json.dumps({
                 "id": test_item_id,
                 "type": "Feature",
@@ -201,9 +198,8 @@ def test_end_to_end_notification_flow():
 
     finally:
         # Cleanup
-        with conn.cursor() as cursor:
+        with db_connection.cursor() as cursor:
             cursor.execute("SELECT pgstac.delete_item(%s);", (test_item_id,))
-        conn.close()
 
 
 def test_k_sink_injection():

.github/workflows/tests/test_pgstac_notifications.py

@@ -12,19 +12,28 @@
 @pytest.fixture(scope='session')
 def db_connection():
     """Create database connection for testing."""
-    # Use the same environment variables as the pgstac bootstrap
+    # Require all database connection parameters to be explicitly set
+    required_vars = ['PGHOST', 'PGPORT', 'PGDATABASE', 'PGUSER', 'PGPASSWORD']
+    missing_vars = [var for var in required_vars if not os.getenv(var)]
+
+    if missing_vars:
+        pytest.fail(f"Required environment variables not set: {', '.join(missing_vars)}")
+
     connection_params = {
-        'host': os.getenv('PGHOST', 'localhost'),
-        'port': int(os.getenv('PGPORT', '5432')),
-        'database': os.getenv('PGDATABASE', 'postgres'),
-        'user': os.getenv('PGUSER', 'postgres'),
-        'password': os.getenv('PGPASSWORD', 'password')
+        'host': os.getenv('PGHOST'),
+        'port': int(os.getenv('PGPORT')),
+        'database': os.getenv('PGDATABASE'),
+        'user': os.getenv('PGUSER'),
+        'password': os.getenv('PGPASSWORD')
     }
 
-    conn = psycopg2.connect(**connection_params)
-    conn.set_isolation_level(psycopg2.extensions.ISOLATION_LEVEL_AUTOCOMMIT)
-    yield conn
-    conn.close()
+    try:
+        conn = psycopg2.connect(**connection_params)
+        conn.set_isolation_level(psycopg2.extensions.ISOLATION_LEVEL_AUTOCOMMIT)
+        yield conn
+        conn.close()
+    except psycopg2.Error as e:
+        pytest.fail(f"Cannot connect to database: {e}")
 
 
 @pytest.fixture(scope='session')