tests again

Author: millerjp

tests/integration/test_fastapi_reconnection_isolation.py

@@ -22,7 +22,7 @@ def _get_cassandra_control(self, container=None):
 
     @pytest.mark.integration
     @pytest.mark.asyncio
-    async def test_session_health_check_pattern(self):
+    async def test_session_health_check_pattern(self, cassandra_container):
         """Test the FastAPI health check pattern that might prevent reconnection."""
         print("\n=== Testing FastAPI Health Check Pattern ===")
 
@@ -71,7 +71,7 @@ async def health_check():
 
             # Disable Cassandra
             print("\nDisabling Cassandra...")
-            control = self._get_cassandra_control()
+            control = self._get_cassandra_control(cassandra_container)
 
             if os.environ.get("CI") == "true":
                 # Still test that health check works with available service
@@ -132,7 +132,7 @@ async def health_check():
 
     @pytest.mark.integration
     @pytest.mark.asyncio
-    async def test_global_session_reconnection(self):
+    async def test_global_session_reconnection(self, cassandra_container):
         """Test reconnection with global session variable like FastAPI."""
         print("\n=== Testing Global Session Reconnection ===")
 
@@ -170,7 +170,7 @@ async def test_global_session_reconnection(self):
             print("✓ Initial query works")
 
             # Get control interface
-            control = self._get_cassandra_control()
+            control = self._get_cassandra_control(cassandra_container)
 
             if os.environ.get("CI") == "true":
                 print("\nSkipping outage simulation in CI")

tests/unit/test_sql_injection_protection.py

@@ -104,8 +104,12 @@ async def test_no_string_interpolation_in_queries(self):
         await mock_session.prepare("SELECT * FROM users LIMIT ?")
         await mock_session.execute(mock_stmt, [int(limit.split(";")[0])])  # Parse safely
 
-        # Verify no direct string interpolation
-        assert all("DROP TABLE" not in str(call) for call in mock_session.execute.call_args_list)
+        # Verify prepared statements were used (not string interpolation)
+        # The execute calls should have the mock statement and parameters, not raw SQL
+        for exec_call in mock_session.execute.call_args_list:
+            # Each call should be execute(mock_stmt, [params])
+            assert exec_call[0][0] == mock_stmt  # First arg is the prepared statement
+            assert isinstance(exec_call[0][1], list)  # Second arg is parameters list
 
     @pytest.mark.asyncio
     async def test_hardcoded_keyspace_names(self):