1+ """
2+ Tests for UUID parameter binding support in psqlpy-sqlalchemy.
3+ """
4+
5+ import asyncio
6+ import uuid
7+ import pytest
8+ from sqlalchemy import Column , Integer , String , create_engine , text
9+ from sqlalchemy .dialects .postgresql import UUID
10+ from sqlalchemy .ext .asyncio import AsyncSession , create_async_engine
11+ from sqlalchemy .orm import DeclarativeBase , sessionmaker
12+ from sqlalchemy .exc import StatementError
13+
14+
15+ import os
16+
17+ # Skip tests if database is not available (check for CI environment or explicit flag)
18+ def should_skip_db_tests ():
19+ """Check if database tests should be skipped."""
20+ # Run tests in CI environment
21+ if os .getenv ('GITHUB_ACTIONS' ):
22+ return False
23+ # Run tests if explicitly enabled
24+ if os .getenv ('RUN_DB_TESTS' ):
25+ return False
26+ # Skip by default in local development
27+ return True
28+
29+ pytestmark = pytest .mark .skipif (
30+ should_skip_db_tests (),
31+ reason = "Database tests require live PostgreSQL connection. Set RUN_DB_TESTS=1 or run in CI."
32+ )
33+
34+
35+ class Base (DeclarativeBase ):
36+ pass
37+
38+
39+ class TestUUIDTable (Base ):
40+ __tablename__ = "test_uuid_table"
41+
42+ id = Column (Integer , primary_key = True )
43+ uid = Column (UUID (as_uuid = True ), nullable = False )
44+ name = Column (String (100 ))
45+
46+
47+ @pytest .fixture
48+ async def engine ():
49+ """Create test engine."""
50+ # Use environment variables for database connection in CI
51+ db_url = os .getenv (
52+ 'DATABASE_URL' ,
53+ 'postgresql+psqlpy://postgres:password@localhost:5432/test_db'
54+ )
55+ engine = create_async_engine (db_url , echo = False )
56+
57+ async with engine .begin () as conn :
58+ await conn .run_sync (Base .metadata .create_all )
59+
60+ yield engine
61+
62+ async with engine .begin () as conn :
63+ await conn .run_sync (Base .metadata .drop_all )
64+
65+ await engine .dispose ()
66+
67+
68+ @pytest .fixture
69+ async def session (engine ):
70+ """Create test session."""
71+ async_session = sessionmaker (engine , class_ = AsyncSession )
72+ async with async_session () as session :
73+ yield session
74+
75+
76+ class TestUUIDParameterBinding :
77+ """Test UUID parameter binding functionality."""
78+
79+ async def test_uuid_object_parameter (self , engine ):
80+ """Test UUID object as parameter."""
81+ test_uuid = uuid .uuid4 ()
82+
83+ async with engine .begin () as conn :
84+ # Insert with UUID object
85+ await conn .execute (
86+ text ("INSERT INTO test_uuid_table (uid, name) VALUES (:uid, :name)" ),
87+ {"uid" : test_uuid , "name" : "test_uuid_object" }
88+ )
89+
90+ # Query with UUID object
91+ result = await conn .execute (
92+ text ("SELECT * FROM test_uuid_table WHERE uid = :uid" ),
93+ {"uid" : test_uuid }
94+ )
95+
96+ rows = result .fetchall ()
97+ assert len (rows ) == 1
98+ assert rows [0 ].name == "test_uuid_object"
99+
100+ async def test_uuid_string_parameter (self , engine ):
101+ """Test UUID string as parameter."""
102+ test_uuid = uuid .uuid4 ()
103+ test_uuid_str = str (test_uuid )
104+
105+ async with engine .begin () as conn :
106+ # Insert with UUID string
107+ await conn .execute (
108+ text ("INSERT INTO test_uuid_table (uid, name) VALUES (:uid, :name)" ),
109+ {"uid" : test_uuid_str , "name" : "test_uuid_string" }
110+ )
111+
112+ # Query with UUID string
113+ result = await conn .execute (
114+ text ("SELECT * FROM test_uuid_table WHERE uid = :uid" ),
115+ {"uid" : test_uuid_str }
116+ )
117+
118+ rows = result .fetchall ()
119+ assert len (rows ) == 1
120+ assert rows [0 ].name == "test_uuid_string"
121+
122+ async def test_uuid_with_explicit_cast (self , engine ):
123+ """Test UUID parameter with explicit PostgreSQL casting."""
124+ test_uuid = uuid .uuid4 ()
125+
126+ async with engine .begin () as conn :
127+ # Insert test data
128+ await conn .execute (
129+ text ("INSERT INTO test_uuid_table (uid, name) VALUES (:uid, :name)" ),
130+ {"uid" : test_uuid , "name" : "test_cast" }
131+ )
132+
133+ # This was the original failing case - explicit UUID casting
134+ result = await conn .execute (
135+ text ("SELECT * FROM test_uuid_table WHERE uid = :uid::UUID LIMIT :limit" ),
136+ {"uid" : str (test_uuid ), "limit" : 2 }
137+ )
138+
139+ rows = result .fetchall ()
140+ assert len (rows ) == 1
141+ assert rows [0 ].name == "test_cast"
142+
143+ async def test_uuid_with_sqlalchemy_orm (self , session ):
144+ """Test UUID with SQLAlchemy ORM."""
145+ test_uuid = uuid .uuid4 ()
146+
147+ # Insert with ORM
148+ test_obj = TestUUIDTable (uid = test_uuid , name = "test_orm" )
149+ session .add (test_obj )
150+ await session .commit ()
151+
152+ # Query with ORM
153+ result = await session .execute (
154+ text ("SELECT * FROM test_uuid_table WHERE uid = :uid ORDER BY id LIMIT :limit" ),
155+ {"uid" : test_uuid , "limit" : 2 }
156+ )
157+
158+ rows = result .fetchall ()
159+ assert len (rows ) == 1
160+ assert rows [0 ].name == "test_orm"
161+
162+ async def test_multiple_uuid_parameters (self , engine ):
163+ """Test multiple UUID parameters in one query."""
164+ uuid1 = uuid .uuid4 ()
165+ uuid2 = uuid .uuid4 ()
166+
167+ async with engine .begin () as conn :
168+ # Insert test data
169+ await conn .execute (
170+ text ("INSERT INTO test_uuid_table (uid, name) VALUES (:uid1, :name1), (:uid2, :name2)" ),
171+ {"uid1" : uuid1 , "name1" : "first" , "uid2" : uuid2 , "name2" : "second" }
172+ )
173+
174+ # Query with multiple UUID parameters
175+ result = await conn .execute (
176+ text ("SELECT * FROM test_uuid_table WHERE uid IN (:uid1, :uid2) ORDER BY name" ),
177+ {"uid1" : uuid1 , "uid2" : uuid2 }
178+ )
179+
180+ rows = result .fetchall ()
181+ assert len (rows ) == 2
182+ assert rows [0 ].name == "first"
183+ assert rows [1 ].name == "second"
184+
185+ async def test_null_uuid_parameter (self , engine ):
186+ """Test NULL UUID parameter."""
187+ async with engine .begin () as conn :
188+ # Query with NULL UUID - should return no results
189+ result = await conn .execute (
190+ text ("SELECT * FROM test_uuid_table WHERE uid = :uid" ),
191+ {"uid" : None }
192+ )
193+
194+ rows = result .fetchall ()
195+ assert len (rows ) == 0
196+
197+ async def test_invalid_uuid_string (self , engine ):
198+ """Test invalid UUID string raises proper error."""
199+ async with engine .begin () as conn :
200+ with pytest .raises ((ValueError , StatementError )):
201+ await conn .execute (
202+ text ("INSERT INTO test_uuid_table (uid, name) VALUES (:uid, :name)" ),
203+ {"uid" : "invalid-uuid-string" , "name" : "test" }
204+ )
205+
206+ async def test_uuid_edge_cases (self , engine ):
207+ """Test UUID edge cases."""
208+ # Test various UUID formats
209+ test_cases = [
210+ uuid .UUID ('00000000-0000-0000-0000-000000000000' ), # Nil UUID
211+ uuid .UUID ('ffffffff-ffff-ffff-ffff-ffffffffffff' ), # Max UUID
212+ uuid .uuid1 (), # Time-based UUID
213+ uuid .uuid4 (), # Random UUID
214+ ]
215+
216+ async with engine .begin () as conn :
217+ for i , test_uuid in enumerate (test_cases ):
218+ await conn .execute (
219+ text ("INSERT INTO test_uuid_table (uid, name) VALUES (:uid, :name)" ),
220+ {"uid" : test_uuid , "name" : f"edge_case_{ i } }
221+ )
222+
223+ # Verify all were inserted correctly
224+ result = await conn .execute (
225+ text ("SELECT COUNT(*) as count FROM test_uuid_table WHERE name LIKE 'edge_case_%'" )
226+ )
227+
228+ count = result .fetchone ().count
229+ assert count == len (test_cases )
230+
231+
232+ class TestUUIDTypeCompatibility :
233+ """Test UUID type compatibility with existing functionality."""
234+
235+ async def test_uuid_column_definition (self , engine ):
236+ """Test that UUID columns are properly defined."""
237+ async with engine .begin () as conn :
238+ # Check table structure
239+ result = await conn .execute (
240+ text ("""
241+ SELECT column_name, data_type
242+ FROM information_schema.columns
243+ WHERE table_name = 'test_uuid_table' AND column_name = 'uid'
244+ """ )
245+ )
246+
247+ row = result .fetchone ()
248+ assert row is not None
249+ assert row .data_type == 'uuid'
250+
251+ async def test_uuid_index_support (self , engine ):
252+ """Test that UUID columns can be indexed."""
253+ async with engine .begin () as conn :
254+ # Create index on UUID column
255+ await conn .execute (
256+ text ("CREATE INDEX IF NOT EXISTS idx_test_uuid_uid ON test_uuid_table(uid)" )
257+ )
258+
259+ # Verify index was created
260+ result = await conn .execute (
261+ text ("""
262+ SELECT indexname
263+ FROM pg_indexes
264+ WHERE tablename = 'test_uuid_table' AND indexname = 'idx_test_uuid_uid'
265+ """ )
266+ )
267+
268+ row = result .fetchone ()
269+ assert row is not None
270+
271+ # Clean up
272+ await conn .execute (text ("DROP INDEX IF EXISTS idx_test_uuid_uid" ))
273+
274+
275+ if __name__ == "__main__" :
276+ # Run tests directly
277+ pytest .main ([__file__ , "-v" ])
0 commit comments