implement UUID parameter binding

Author: Eugene Shershen

tests/test_uuid_support.py

@@ -2,33 +2,33 @@
 Tests for UUID parameter binding support in psqlpy-sqlalchemy.
 """
 
-import asyncio
+import os
 import uuid
+
 import pytest
-from sqlalchemy import Column, Integer, String, create_engine, text
+from sqlalchemy import Column, Integer, String, text
 from sqlalchemy.dialects.postgresql import UUID
+from sqlalchemy.exc import StatementError
 from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
 from sqlalchemy.orm import DeclarativeBase, sessionmaker
-from sqlalchemy.exc import StatementError
-
 
-import os
 
 # Skip tests if database is not available (check for CI environment or explicit flag)
 def should_skip_db_tests():
     """Check if database tests should be skipped."""
     # Run tests in CI environment
-    if os.getenv('GITHUB_ACTIONS'):
+    if os.getenv("GITHUB_ACTIONS"):
         return False
     # Run tests if explicitly enabled
-    if os.getenv('RUN_DB_TESTS'):
+    if os.getenv("RUN_DB_TESTS"):
         return False
     # Skip by default in local development
     return True
 
+
 pytestmark = pytest.mark.skipif(
-    should_skip_db_tests(), 
-    reason="Database tests require live PostgreSQL connection. Set RUN_DB_TESTS=1 or run in CI."
+    should_skip_db_tests(),
+    reason="Database tests require live PostgreSQL connection. Set RUN_DB_TESTS=1 or run in CI.",
 )
 
 
@@ -38,7 +38,7 @@ class Base(DeclarativeBase):
 
 class TestUUIDTable(Base):
     __tablename__ = "test_uuid_table"
-    
+
     id = Column(Integer, primary_key=True)
     uid = Column(UUID(as_uuid=True), nullable=False)
     name = Column(String(100))
@@ -49,19 +49,19 @@ async def engine():
     """Create test engine."""
     # Use environment variables for database connection in CI
     db_url = os.getenv(
-        'DATABASE_URL', 
-        'postgresql+psqlpy://postgres:password@localhost:5432/test_db'
+        "DATABASE_URL",
+        "postgresql+psqlpy://postgres:password@localhost:5432/test_db",
     )
     engine = create_async_engine(db_url, echo=False)
-    
+
     async with engine.begin() as conn:
         await conn.run_sync(Base.metadata.create_all)
-    
+
     yield engine
-    
+
     async with engine.begin() as conn:
         await conn.run_sync(Base.metadata.drop_all)
-    
+
     await engine.dispose()
 
 
@@ -75,203 +75,230 @@ async def session(engine):
 
 class TestUUIDParameterBinding:
     """Test UUID parameter binding functionality."""
-    
+
     async def test_uuid_object_parameter(self, engine):
         """Test UUID object as parameter."""
         test_uuid = uuid.uuid4()
-        
+
         async with engine.begin() as conn:
             # Insert with UUID object
             await conn.execute(
-                text("INSERT INTO test_uuid_table (uid, name) VALUES (:uid, :name)"),
-                {"uid": test_uuid, "name": "test_uuid_object"}
+                text(
+                    "INSERT INTO test_uuid_table (uid, name) VALUES (:uid, :name)"
+                ),
+                {"uid": test_uuid, "name": "test_uuid_object"},
             )
-            
+
             # Query with UUID object
             result = await conn.execute(
                 text("SELECT * FROM test_uuid_table WHERE uid = :uid"),
-                {"uid": test_uuid}
+                {"uid": test_uuid},
             )
-            
+
             rows = result.fetchall()
             assert len(rows) == 1
             assert rows[0].name == "test_uuid_object"
-    
+
     async def test_uuid_string_parameter(self, engine):
         """Test UUID string as parameter."""
         test_uuid = uuid.uuid4()
         test_uuid_str = str(test_uuid)
-        
+
         async with engine.begin() as conn:
             # Insert with UUID string
             await conn.execute(
-                text("INSERT INTO test_uuid_table (uid, name) VALUES (:uid, :name)"),
-                {"uid": test_uuid_str, "name": "test_uuid_string"}
+                text(
+                    "INSERT INTO test_uuid_table (uid, name) VALUES (:uid, :name)"
+                ),
+                {"uid": test_uuid_str, "name": "test_uuid_string"},
             )
-            
+
             # Query with UUID string
             result = await conn.execute(
                 text("SELECT * FROM test_uuid_table WHERE uid = :uid"),
-                {"uid": test_uuid_str}
+                {"uid": test_uuid_str},
             )
-            
+
             rows = result.fetchall()
             assert len(rows) == 1
             assert rows[0].name == "test_uuid_string"
-    
+
     async def test_uuid_with_explicit_cast(self, engine):
         """Test UUID parameter with explicit PostgreSQL casting."""
         test_uuid = uuid.uuid4()
-        
+
         async with engine.begin() as conn:
             # Insert test data
             await conn.execute(
-                text("INSERT INTO test_uuid_table (uid, name) VALUES (:uid, :name)"),
-                {"uid": test_uuid, "name": "test_cast"}
+                text(
+                    "INSERT INTO test_uuid_table (uid, name) VALUES (:uid, :name)"
+                ),
+                {"uid": test_uuid, "name": "test_cast"},
             )
-            
+
             # This was the original failing case - explicit UUID casting
             result = await conn.execute(
-                text("SELECT * FROM test_uuid_table WHERE uid = :uid::UUID LIMIT :limit"),
-                {"uid": str(test_uuid), "limit": 2}
+                text(
+                    "SELECT * FROM test_uuid_table WHERE uid = :uid::UUID LIMIT :limit"
+                ),
+                {"uid": str(test_uuid), "limit": 2},
             )
-            
+
             rows = result.fetchall()
             assert len(rows) == 1
             assert rows[0].name == "test_cast"
-    
+
     async def test_uuid_with_sqlalchemy_orm(self, session):
         """Test UUID with SQLAlchemy ORM."""
         test_uuid = uuid.uuid4()
-        
+
         # Insert with ORM
         test_obj = TestUUIDTable(uid=test_uuid, name="test_orm")
         session.add(test_obj)
         await session.commit()
-        
+
         # Query with ORM
         result = await session.execute(
-            text("SELECT * FROM test_uuid_table WHERE uid = :uid ORDER BY id LIMIT :limit"),
-            {"uid": test_uuid, "limit": 2}
+            text(
+                "SELECT * FROM test_uuid_table WHERE uid = :uid ORDER BY id LIMIT :limit"
+            ),
+            {"uid": test_uuid, "limit": 2},
         )
-        
+
         rows = result.fetchall()
         assert len(rows) == 1
         assert rows[0].name == "test_orm"
-    
+
     async def test_multiple_uuid_parameters(self, engine):
         """Test multiple UUID parameters in one query."""
         uuid1 = uuid.uuid4()
         uuid2 = uuid.uuid4()
-        
+
         async with engine.begin() as conn:
             # Insert test data
             await conn.execute(
-                text("INSERT INTO test_uuid_table (uid, name) VALUES (:uid1, :name1), (:uid2, :name2)"),
-                {"uid1": uuid1, "name1": "first", "uid2": uuid2, "name2": "second"}
+                text(
+                    "INSERT INTO test_uuid_table (uid, name) VALUES (:uid1, :name1), (:uid2, :name2)"
+                ),
+                {
+                    "uid1": uuid1,
+                    "name1": "first",
+                    "uid2": uuid2,
+                    "name2": "second",
+                },
             )
-            
+
             # Query with multiple UUID parameters
             result = await conn.execute(
-                text("SELECT * FROM test_uuid_table WHERE uid IN (:uid1, :uid2) ORDER BY name"),
-                {"uid1": uuid1, "uid2": uuid2}
+                text(
+                    "SELECT * FROM test_uuid_table WHERE uid IN (:uid1, :uid2) ORDER BY name"
+                ),
+                {"uid1": uuid1, "uid2": uuid2},
             )
-            
+
             rows = result.fetchall()
             assert len(rows) == 2
             assert rows[0].name == "first"
             assert rows[1].name == "second"
-    
+
     async def test_null_uuid_parameter(self, engine):
         """Test NULL UUID parameter."""
         async with engine.begin() as conn:
             # Query with NULL UUID - should return no results
             result = await conn.execute(
                 text("SELECT * FROM test_uuid_table WHERE uid = :uid"),
-                {"uid": None}
+                {"uid": None},
             )
-            
+
             rows = result.fetchall()
             assert len(rows) == 0
-    
+
     async def test_invalid_uuid_string(self, engine):
         """Test invalid UUID string raises proper error."""
         async with engine.begin() as conn:
             with pytest.raises((ValueError, StatementError)):
                 await conn.execute(
-                    text("INSERT INTO test_uuid_table (uid, name) VALUES (:uid, :name)"),
-                    {"uid": "invalid-uuid-string", "name": "test"}
+                    text(
+                        "INSERT INTO test_uuid_table (uid, name) VALUES (:uid, :name)"
+                    ),
+                    {"uid": "invalid-uuid-string", "name": "test"},
                 )
-    
+
     async def test_uuid_edge_cases(self, engine):
         """Test UUID edge cases."""
         # Test various UUID formats
         test_cases = [
-            uuid.UUID('00000000-0000-0000-0000-000000000000'),  # Nil UUID
-            uuid.UUID('ffffffff-ffff-ffff-ffff-ffffffffffff'),  # Max UUID
+            uuid.UUID("00000000-0000-0000-0000-000000000000"),  # Nil UUID
+            uuid.UUID("ffffffff-ffff-ffff-ffff-ffffffffffff"),  # Max UUID
             uuid.uuid1(),  # Time-based UUID
             uuid.uuid4(),  # Random UUID
         ]
-        
+
         async with engine.begin() as conn:
             for i, test_uuid in enumerate(test_cases):
                 await conn.execute(
-                    text("INSERT INTO test_uuid_table (uid, name) VALUES (:uid, :name)"),
-                    {"uid": test_uuid, "name": f"edge_case_{i}"}
+                    text(
+                        "INSERT INTO test_uuid_table (uid, name) VALUES (:uid, :name)"
+                    ),
+                    {"uid": test_uuid, "name": f"edge_case_{i}"},
                 )
-            
+
             # Verify all were inserted correctly
             result = await conn.execute(
-                text("SELECT COUNT(*) as count FROM test_uuid_table WHERE name LIKE 'edge_case_%'")
+                text(
+                    "SELECT COUNT(*) as count FROM test_uuid_table WHERE name LIKE 'edge_case_%'"
+                )
             )
-            
+
             count = result.fetchone().count
             assert count == len(test_cases)
 
 
 class TestUUIDTypeCompatibility:
     """Test UUID type compatibility with existing functionality."""
-    
+
     async def test_uuid_column_definition(self, engine):
         """Test that UUID columns are properly defined."""
         async with engine.begin() as conn:
             # Check table structure
             result = await conn.execute(
                 text("""
-                SELECT column_name, data_type 
-                FROM information_schema.columns 
+                SELECT column_name, data_type
+                FROM information_schema.columns
                 WHERE table_name = 'test_uuid_table' AND column_name = 'uid'
                 """)
             )
-            
+
             row = result.fetchone()
             assert row is not None
-            assert row.data_type == 'uuid'
-    
+            assert row.data_type == "uuid"
+
     async def test_uuid_index_support(self, engine):
         """Test that UUID columns can be indexed."""
         async with engine.begin() as conn:
             # Create index on UUID column
             await conn.execute(
-                text("CREATE INDEX IF NOT EXISTS idx_test_uuid_uid ON test_uuid_table(uid)")
+                text(
+                    "CREATE INDEX IF NOT EXISTS idx_test_uuid_uid ON test_uuid_table(uid)"
+                )
             )
-            
+
             # Verify index was created
             result = await conn.execute(
                 text("""
-                SELECT indexname 
-                FROM pg_indexes 
+                SELECT indexname
+                FROM pg_indexes
                 WHERE tablename = 'test_uuid_table' AND indexname = 'idx_test_uuid_uid'
                 """)
             )
-            
+
             row = result.fetchone()
             assert row is not None
-            
+
             # Clean up
             await conn.execute(text("DROP INDEX IF EXISTS idx_test_uuid_uid"))
 
 
 if __name__ == "__main__":
     # Run tests directly
-    pytest.main([__file__, "-v"])
+    pytest.main([__file__, "-v"])