Skip to content

Commit 10fb51b

Browse files
sakthivelmaniisakthivelmanii
committed
test(sqlalchemy-spanner): add mockserver 15-matrix tests for types.Uuid vs types.UUID
1 parent b60699e commit 10fb51b

1 file changed

Lines changed: 192 additions & 17 deletions

File tree

  • packages/sqlalchemy-spanner/tests/mockserver_tests

‎packages/sqlalchemy-spanner/tests/mockserver_tests/test_uuid.py‎

Lines changed: 192 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -12,17 +12,47 @@
1212
# See the License for the specific language governing permissions and
1313
# limitations under the License.
1414

15-
from sqlalchemy import Column, MetaData, Table, types
15+
import uuid
16+
17+
from google.cloud.spanner_v1 import TypeCode
18+
from sqlalchemy import Column, MetaData, Table, select, types
19+
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column
1620
from sqlalchemy.schema import CreateTable
17-
from sqlalchemy.testing import eq_, fixtures
21+
from sqlalchemy.testing import eq_
22+
1823
from google.cloud.sqlalchemy_spanner.sqlalchemy_spanner import (
1924
SpannerDialect,
2025
_type_map,
2126
_type_map_inv,
2227
)
28+
from tests.mockserver_tests.mock_server_test_base import (
29+
MockServerTestBase,
30+
add_single_result,
31+
)
32+
33+
34+
class Base(DeclarativeBase):
35+
pass
36+
37+
38+
class UserUuid(Base):
39+
__tablename__ = "users_uuid"
40+
id: Mapped[uuid.UUID] = mapped_column(types.Uuid, primary_key=True)
41+
42+
43+
class UserUUID(Base):
44+
__tablename__ = "users_UUID"
45+
id: Mapped[uuid.UUID] = mapped_column(types.UUID, primary_key=True)
46+
2347

48+
class UserUuidNativeFalse(Base):
49+
__tablename__ = "users_uuid_native_false"
50+
id: Mapped[uuid.UUID] = mapped_column(
51+
types.Uuid(native_uuid=False), primary_key=True
52+
)
2453

25-
class UuidTest(fixtures.TestBase):
54+
55+
class TestUuidMockServer(MockServerTestBase):
2656
def test_uuid_type_mapping(self):
2757
"""Test UUID is registered in _type_map and _type_map_inv."""
2858
assert "UUID" in _type_map
@@ -70,21 +100,166 @@ def test_uuid_ddl_compilation_native_enabled(self):
70100
statement = str(CreateTable(table).compile(dialect=dialect)).strip()
71101
assert "user_id UUID NOT NULL" in statement
72102

73-
def test_uuid_python_conversion_legacy(self):
74-
"""Test that types.Uuid automatically converts uuid.UUID to/from str
75-
when native_uuid is False.
103+
def test_uuid_ddl_compilation_native_false_override(self):
104+
"""Test DDL compilation emits STRING(36) for types.Uuid(native_uuid=False) even
105+
when supports_native_uuid is set to True on dialect.
76106
"""
77-
import uuid
78-
79107
dialect = SpannerDialect()
80-
dialect.supports_native_uuid = False
81-
uuid_type = types.Uuid()
82-
test_uuid = uuid.UUID("123e4567-e89b-12d3-a456-426614174000")
108+
dialect.supports_native_uuid = True
109+
metadata = MetaData()
110+
table = Table(
111+
"test_uuid_table",
112+
metadata,
113+
Column("user_id", types.Uuid(native_uuid=False), primary_key=True),
114+
)
115+
statement = str(CreateTable(table).compile(dialect=dialect)).strip()
116+
assert "user_id STRING(36) NOT NULL" in statement
117+
118+
# -----------------------------------------------------------------
119+
# 1. Flag is disabled (supports_native_uuid = False)
120+
# -----------------------------------------------------------------
121+
def test_1_1_flag_disabled_uuid_db_string(self):
122+
"""1.1 Flag disabled + mapped Uuid + DB STRING(36) -> user.id is uuid.UUID"""
123+
engine = self.create_engine()
124+
engine.dialect.supports_native_uuid = False
125+
sql = "SELECT users_uuid.id\nFROM users_uuid"
126+
raw_uuid_str = "550e8400-e29b-41d4-a716-446655440000"
127+
add_single_result(sql, "id", TypeCode.STRING, [(raw_uuid_str,)])
83128

84-
# Test bind processor converts uuid.UUID -> str
85-
bind_proc = uuid_type.bind_processor(dialect)
86-
eq_(bind_proc(test_uuid), "123e4567e89b12d3a456426614174000")
129+
with Session(engine) as session:
130+
user_id = session.scalars(select(UserUuid.id)).first()
131+
eq_(type(user_id), uuid.UUID)
132+
eq_(user_id, uuid.UUID(raw_uuid_str))
133+
134+
def test_1_2_flag_disabled_uuid_db_uuid(self):
135+
"""1.2 Flag disabled + mapped Uuid + DB UUID -> AttributeError"""
136+
engine = self.create_engine()
137+
engine.dialect.supports_native_uuid = False
138+
sql = "SELECT users_uuid.id\nFROM users_uuid"
139+
raw_uuid_str = "550e8400-e29b-41d4-a716-446655440000"
140+
add_single_result(sql, "id", TypeCode.UUID, [(raw_uuid_str,)])
141+
142+
with Session(engine) as session:
143+
try:
144+
_ = session.scalars(select(UserUuid.id)).first()
145+
assert False, "Expected AttributeError"
146+
except AttributeError:
147+
pass
148+
149+
def test_1_3_flag_disabled_UUID_db_string(self):
150+
"""1.3 Flag disabled + mapped UUID (all-caps) + DB STRING(36) -> uuid.UUID"""
151+
engine = self.create_engine()
152+
engine.dialect.supports_native_uuid = False
153+
sql = "SELECT `users_UUID`.id\nFROM `users_UUID`"
154+
raw_uuid_str = "550e8400-e29b-41d4-a716-446655440000"
155+
add_single_result(sql, "id", TypeCode.STRING, [(raw_uuid_str,)])
156+
157+
with Session(engine) as session:
158+
user_id = session.scalars(select(UserUUID.id)).first()
159+
eq_(type(user_id), uuid.UUID)
160+
eq_(user_id, uuid.UUID(raw_uuid_str))
161+
162+
def test_1_4_flag_disabled_UUID_db_uuid(self):
163+
"""1.4 Flag disabled + mapped UUID (all-caps) + DB UUID -> AttributeError"""
164+
engine = self.create_engine()
165+
engine.dialect.supports_native_uuid = False
166+
sql = "SELECT `users_UUID`.id\nFROM `users_UUID`"
167+
raw_uuid_str = "550e8400-e29b-41d4-a716-446655440000"
168+
add_single_result(sql, "id", TypeCode.UUID, [(raw_uuid_str,)])
169+
170+
with Session(engine) as session:
171+
try:
172+
_ = session.scalars(select(UserUUID.id)).first()
173+
assert False, "Expected AttributeError"
174+
except AttributeError:
175+
pass
176+
177+
# -----------------------------------------------------------------
178+
# 2. Flag is enabled (supports_native_uuid = True)
179+
# -----------------------------------------------------------------
180+
def test_2_1_flag_enabled_uuid_db_string(self):
181+
"""2.1 Flag enabled + mapped Uuid + DB STRING(36) -> user.id is str"""
182+
engine = self.create_engine()
183+
engine.dialect.supports_native_uuid = True
184+
sql = "SELECT users_uuid.id\nFROM users_uuid"
185+
raw_uuid_str = "550e8400-e29b-41d4-a716-446655440000"
186+
add_single_result(sql, "id", TypeCode.STRING, [(raw_uuid_str,)])
187+
188+
with Session(engine) as session:
189+
user_id = session.scalars(select(UserUuid.id)).first()
190+
eq_(type(user_id), str)
191+
eq_(user_id, raw_uuid_str)
192+
193+
def test_2_2_flag_enabled_uuid_db_uuid(self):
194+
"""2.2 Flag enabled + mapped Uuid + DB UUID -> user.id is uuid.UUID"""
195+
engine = self.create_engine()
196+
engine.dialect.supports_native_uuid = True
197+
sql = "SELECT users_uuid.id\nFROM users_uuid"
198+
raw_uuid_str = "550e8400-e29b-41d4-a716-446655440000"
199+
add_single_result(sql, "id", TypeCode.UUID, [(raw_uuid_str,)])
200+
201+
with Session(engine) as session:
202+
user_id = session.scalars(select(UserUuid.id)).first()
203+
eq_(type(user_id), uuid.UUID)
204+
eq_(user_id, uuid.UUID(raw_uuid_str))
205+
206+
def test_2_3_flag_enabled_UUID_db_string(self):
207+
"""2.3 Flag enabled + mapped UUID (all-caps) + DB STRING(36) -> str"""
208+
engine = self.create_engine()
209+
engine.dialect.supports_native_uuid = True
210+
sql = "SELECT `users_UUID`.id\nFROM `users_UUID`"
211+
raw_uuid_str = "550e8400-e29b-41d4-a716-446655440000"
212+
add_single_result(sql, "id", TypeCode.STRING, [(raw_uuid_str,)])
213+
214+
with Session(engine) as session:
215+
user_id = session.scalars(select(UserUUID.id)).first()
216+
eq_(type(user_id), str)
217+
eq_(user_id, raw_uuid_str)
218+
219+
def test_2_4_flag_enabled_UUID_db_uuid(self):
220+
"""2.4 Flag enabled + mapped UUID (all-caps) + DB UUID -> uuid.UUID"""
221+
engine = self.create_engine()
222+
engine.dialect.supports_native_uuid = True
223+
sql = "SELECT `users_UUID`.id\nFROM `users_UUID`"
224+
raw_uuid_str = "550e8400-e29b-41d4-a716-446655440000"
225+
add_single_result(sql, "id", TypeCode.UUID, [(raw_uuid_str,)])
226+
227+
with Session(engine) as session:
228+
user_id = session.scalars(select(UserUUID.id)).first()
229+
eq_(type(user_id), uuid.UUID)
230+
eq_(user_id, uuid.UUID(raw_uuid_str))
231+
232+
# -----------------------------------------------------------------
233+
# 3. Explicit native_uuid = False override
234+
# -----------------------------------------------------------------
235+
def test_3_1_flag_enabled_native_false_override_db_string(self):
236+
"""3.1 Flag enabled + types.Uuid(native_uuid=False) + DB STRING(36)
237+
-> user.id is uuid.UUID (override forces conversion)
238+
"""
239+
engine = self.create_engine()
240+
engine.dialect.supports_native_uuid = True
241+
sql = "SELECT users_uuid_native_false.id\nFROM users_uuid_native_false"
242+
raw_uuid_str = "550e8400-e29b-41d4-a716-446655440000"
243+
add_single_result(sql, "id", TypeCode.STRING, [(raw_uuid_str,)])
244+
245+
with Session(engine) as session:
246+
user_id = session.scalars(select(UserUuidNativeFalse.id)).first()
247+
eq_(type(user_id), uuid.UUID)
248+
eq_(user_id, uuid.UUID(raw_uuid_str))
249+
250+
def test_3_2_flag_enabled_native_false_override_db_uuid(self):
251+
"""3.2 Flag enabled + types.Uuid(native_uuid=False) + DB UUID
252+
-> AttributeError (override forces conversion, failing on native UUID)
253+
"""
254+
engine = self.create_engine()
255+
engine.dialect.supports_native_uuid = True
256+
sql = "SELECT users_uuid_native_false.id\nFROM users_uuid_native_false"
257+
raw_uuid_str = "550e8400-e29b-41d4-a716-446655440000"
258+
add_single_result(sql, "id", TypeCode.UUID, [(raw_uuid_str,)])
87259

88-
# Test result processor converts str -> uuid.UUID
89-
res_proc = uuid_type.result_processor(dialect, "STRING")
90-
eq_(res_proc("123e4567-e89b-12d3-a456-426614174000"), test_uuid)
260+
with Session(engine) as session:
261+
try:
262+
_ = session.scalars(select(UserUuidNativeFalse.id)).first()
263+
assert False, "Expected AttributeError"
264+
except AttributeError:
265+
pass

0 commit comments

Comments
 (0)