Skip to content
This repository was archived by the owner on Apr 15, 2025. It is now read-only.

Commit 18b364c

Browse files
jonathanbladejonathanblade
committed
chore(tests): fix transaction isolation level tests
1 parent e2f70cd commit 18b364c

File tree

4 files changed

+122
-242
lines changed

4 files changed

+122
-242
lines changed

‎databases/sync_tests/test_transactions.py‎

Lines changed: 53 additions & 117 deletions
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,7 @@
88
from prisma import Prisma
99
from prisma.models import User, Profile
1010

11-
from ..utils import CURRENT_DATABASE
11+
from ..utils import CURRENT_DATABASE, RawQueries
1212

1313

1414
def test_model_query(client: Prisma) -> None:
@@ -203,122 +203,58 @@ def test_transaction_already_closed(client: Prisma) -> None:
203203
assert exc.match('Transaction already closed')
204204

205205

206-
@pytest.mark.skipif(CURRENT_DATABASE in ['cockroachdb', 'sqlite'], reason='Not available')
207-
def test_read_uncommited_isolation_level(client: Prisma) -> None:
208-
"""A transaction isolation level is set to `READ_UNCOMMITED`"""
209-
client2 = Prisma()
210-
client2.connect()
211-
212-
user = client.user.create(data={'name': 'Robert'})
213-
214-
with client.tx(isolation_level=prisma.TransactionIsolationLevel.READ_UNCOMMITED) as tx1:
215-
tx1_user = tx1.user.find_first_or_raise(where={'id': user.id})
216-
tx1_count = tx1.user.count()
217-
218-
with client2.tx() as tx2:
219-
tx2.user.update(data={'name': 'Tegan'}, where={'id': user.id})
220-
tx2.user.create(data={'name': 'Bobby'})
221-
222-
dirty_user = tx1.user.find_first_or_raise(where={'id': user.id})
223-
224-
non_repeatable_user = tx1.user.find_first_or_raise(where={'id': user.id})
225-
phantom_count = tx1.user.count()
226-
227-
# Have dirty read
228-
assert tx1_user.name != dirty_user.name
229-
# Have non-repeatable read
230-
assert tx1_user.name != non_repeatable_user.name
231-
# Have phantom read
232-
assert tx1_count != phantom_count
233-
234-
235-
@pytest.mark.skipif(CURRENT_DATABASE in ['cockroachdb', 'sqlite'], reason='Not available')
236-
def test_read_commited_isolation_level(client: Prisma) -> None:
237-
"""A transaction isolation level is set to `READ_COMMITED`"""
238-
client2 = Prisma()
239-
client2.connect()
240-
241-
user = client.user.create(data={'name': 'Robert'})
242-
243-
with client.tx(isolation_level=prisma.TransactionIsolationLevel.READ_COMMITED) as tx1:
244-
tx1_user = tx1.user.find_first_or_raise(where={'id': user.id})
245-
tx1_count = tx1.user.count()
246-
247-
with client2.tx() as tx2:
248-
tx2.user.update(data={'name': 'Tegan'}, where={'id': user.id})
249-
tx2.user.create(data={'name': 'Bobby'})
250-
251-
dirty_user = tx1.user.find_first_or_raise(where={'id': user.id})
252-
253-
non_repeatable_user = tx1.user.find_first_or_raise(where={'id': user.id})
254-
phantom_count = tx1.user.count()
255-
256-
# No dirty read
257-
assert tx1_user.name == dirty_user.name
258-
# Have non-repeatable read
259-
assert tx1_user.name != non_repeatable_user.name
260-
# Have phantom read
261-
assert tx1_count != phantom_count
262-
263-
264-
@pytest.mark.skipif(CURRENT_DATABASE in ['cockroachdb', 'sqlite'], reason='Not available')
265-
def test_repeatable_read_isolation_level(client: Prisma) -> None:
266-
"""A transaction isolation level is set to `REPEATABLE_READ`"""
267-
client2 = Prisma()
268-
client2.connect()
269-
270-
user = client.user.create(data={'name': 'Robert'})
271-
272-
with client.tx(isolation_level=prisma.TransactionIsolationLevel.REPEATABLE_READ) as tx1:
273-
tx1_user = tx1.user.find_first_or_raise(where={'id': user.id})
274-
tx1_count = tx1.user.count()
275-
276-
with client2.tx() as tx2:
277-
tx2.user.update(data={'name': 'Tegan'}, where={'id': user.id})
278-
tx2.user.create(data={'name': 'Bobby'})
279-
280-
dirty_user = tx1.user.find_first_or_raise(where={'id': user.id})
281-
282-
non_repeatable_user = tx1.user.find_first_or_raise(where={'id': user.id})
283-
phantom_count = tx1.user.count()
284-
285-
# No dirty read
286-
assert tx1_user.name == dirty_user.name
287-
# No non-repeatable read
288-
assert tx1_user.name == non_repeatable_user.name
289-
# Have phantom read
290-
assert tx1_count != phantom_count
291-
292-
293-
@pytest.mark.skipif(True, reason='Available for SQL Server only')
294-
def test_snapshot_isolation_level() -> None:
295-
"""A transaction isolation level is set to `SNAPSHOT`"""
296-
raise NotImplementedError
297-
298-
299-
def test_serializable_isolation_level(client: Prisma) -> None:
300-
"""A transaction isolation level is set to `SERIALIZABLE`"""
301-
client2 = Prisma()
302-
client2.connect()
303-
304-
user = client.user.create(data={'name': 'Robert'})
305-
306-
with client.tx(isolation_level=prisma.TransactionIsolationLevel.SERIALIZABLE) as tx1:
307-
tx1_user = tx1.user.find_first_or_raise(where={'id': user.id})
308-
tx1_count = tx1.user.count()
309-
310-
with client2.tx() as tx2:
311-
tx2.user.update(data={'name': 'Tegan'}, where={'id': user.id})
312-
tx2.user.create(data={'name': 'Bobby'})
206+
@pytest.mark.parametrize(
207+
('input_level', 'expected_level'),
208+
[
209+
pytest.param(
210+
prisma.TransactionIsolationLevel.READ_UNCOMMITTED,
211+
'READ_UNCOMMITTED',
212+
id='read uncommitted',
213+
marks=pytest.mark.skipif(CURRENT_DATABASE in ['cockroachdb', 'sqlite'], reason='Not available'),
214+
),
215+
pytest.param(
216+
prisma.TransactionIsolationLevel.READ_COMMITTED,
217+
'READ_COMMITTED',
218+
id='read committed',
219+
marks=pytest.mark.skipif(CURRENT_DATABASE in ['cockroachdb', 'sqlite'], reason='Not available'),
220+
),
221+
pytest.param(
222+
prisma.TransactionIsolationLevel.REPEATABLE_READ,
223+
'REPEATABLE_READ',
224+
id='repeatable read',
225+
marks=pytest.mark.skipif(CURRENT_DATABASE in ['cockroachdb', 'sqlite'], reason='Not available'),
226+
),
227+
pytest.param(
228+
prisma.TransactionIsolationLevel.SNAPSHOT,
229+
'SNAPSHOT',
230+
id='snapshot',
231+
marks=pytest.mark.skipif(True, reason='Available for SQL Server only'),
232+
),
233+
pytest.param(
234+
prisma.TransactionIsolationLevel.SERIALIZABLE,
235+
'SERIALIZABLE',
236+
id='serializable',
237+
marks=pytest.mark.skipif(
238+
CURRENT_DATABASE == 'sqlite', reason='PRAGMA has only effect in shared-cache mode'
239+
),
240+
),
241+
],
242+
)
243+
# TODO: remove after issue will be resolved
244+
@pytest.mark.skipif(CURRENT_DATABASE in ['mysql', 'mariadb'], reason='https://github.com/prisma/prisma/issues/22890')
245+
def test_isolation_level(
246+
client: Prisma, raw_queries: RawQueries, input_level: prisma.TransactionIsolationLevel, expected_level: str
247+
) -> None:
248+
"""A transaction isolation level is set correctly"""
249+
with client.tx(isolation_level=input_level) as tx:
250+
results = tx.query_raw(raw_queries.select_tx_isolation)
313251

314-
dirty_user = tx1.user.find_first_or_raise(where={'id': user.id})
252+
assert len(results) == 1
315253

316-
non_repeatable_user = tx1.user.find_first_or_raise(where={'id': user.id})
317-
phantom_count = tx1.user.count()
254+
row = results[0]
255+
assert any(row)
318256

319-
# No dirty read
320-
assert tx1_user.name == dirty_user.name
321-
# No non-repeatable read
322-
assert tx1_user.name == non_repeatable_user.name
323-
# No phantom read
324-
assert tx1_count == phantom_count
257+
level = next(iter(row.values()))
258+
# The result can depends on the database, so we do upper() and replace()
259+
level = str(level).upper().replace(' ', '_').replace('-', '_')
260+
assert level == expected_level

‎databases/tests/test_transactions.py‎

Lines changed: 53 additions & 121 deletions
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,7 @@
88
from prisma import Prisma
99
from prisma.models import User, Profile
1010

11-
from ..utils import CURRENT_DATABASE
11+
from ..utils import CURRENT_DATABASE, RawQueries
1212

1313

1414
@pytest.mark.asyncio
@@ -215,126 +215,58 @@ async def test_transaction_already_closed(client: Prisma) -> None:
215215

216216

217217
@pytest.mark.asyncio
218-
@pytest.mark.skipif(CURRENT_DATABASE in ['cockroachdb', 'sqlite'], reason='Not available')
219-
async def test_read_uncommited_isolation_level(client: Prisma) -> None:
220-
"""A transaction isolation level is set to `READ_UNCOMMITED`"""
221-
client2 = Prisma()
222-
await client2.connect()
223-
224-
user = await client.user.create(data={'name': 'Robert'})
225-
226-
async with client.tx(isolation_level=prisma.TransactionIsolationLevel.READ_UNCOMMITED) as tx1:
227-
tx1_user = await tx1.user.find_first_or_raise(where={'id': user.id})
228-
tx1_count = await tx1.user.count()
229-
230-
async with client2.tx() as tx2:
231-
await tx2.user.update(data={'name': 'Tegan'}, where={'id': user.id})
232-
await tx2.user.create(data={'name': 'Bobby'})
233-
234-
dirty_user = await tx1.user.find_first_or_raise(where={'id': user.id})
235-
236-
non_repeatable_user = await tx1.user.find_first_or_raise(where={'id': user.id})
237-
phantom_count = await tx1.user.count()
238-
239-
# Have dirty read
240-
assert tx1_user.name != dirty_user.name
241-
# Have non-repeatable read
242-
assert tx1_user.name != non_repeatable_user.name
243-
# Have phantom read
244-
assert tx1_count != phantom_count
245-
246-
247-
@pytest.mark.asyncio
248-
@pytest.mark.skipif(CURRENT_DATABASE in ['cockroachdb', 'sqlite'], reason='Not available')
249-
async def test_read_commited_isolation_level(client: Prisma) -> None:
250-
"""A transaction isolation level is set to `READ_COMMITED`"""
251-
client2 = Prisma()
252-
await client2.connect()
253-
254-
user = await client.user.create(data={'name': 'Robert'})
255-
256-
async with client.tx(isolation_level=prisma.TransactionIsolationLevel.READ_COMMITED) as tx1:
257-
tx1_user = await tx1.user.find_first_or_raise(where={'id': user.id})
258-
tx1_count = await tx1.user.count()
259-
260-
async with client2.tx() as tx2:
261-
await tx2.user.update(data={'name': 'Tegan'}, where={'id': user.id})
262-
await tx2.user.create(data={'name': 'Bobby'})
263-
264-
dirty_user = await tx1.user.find_first_or_raise(where={'id': user.id})
265-
266-
non_repeatable_user = await tx1.user.find_first_or_raise(where={'id': user.id})
267-
phantom_count = await tx1.user.count()
268-
269-
# No dirty read
270-
assert tx1_user.name == dirty_user.name
271-
# Have non-repeatable read
272-
assert tx1_user.name != non_repeatable_user.name
273-
# Have phantom read
274-
assert tx1_count != phantom_count
275-
276-
277-
@pytest.mark.asyncio
278-
@pytest.mark.skipif(CURRENT_DATABASE in ['cockroachdb', 'sqlite'], reason='Not available')
279-
async def test_repeatable_read_isolation_level(client: Prisma) -> None:
280-
"""A transaction isolation level is set to `REPEATABLE_READ`"""
281-
client2 = Prisma()
282-
await client2.connect()
283-
284-
user = await client.user.create(data={'name': 'Robert'})
285-
286-
async with client.tx(isolation_level=prisma.TransactionIsolationLevel.REPEATABLE_READ) as tx1:
287-
tx1_user = await tx1.user.find_first_or_raise(where={'id': user.id})
288-
tx1_count = await tx1.user.count()
289-
290-
async with client2.tx() as tx2:
291-
await tx2.user.update(data={'name': 'Tegan'}, where={'id': user.id})
292-
await tx2.user.create(data={'name': 'Bobby'})
293-
294-
dirty_user = await tx1.user.find_first_or_raise(where={'id': user.id})
295-
296-
non_repeatable_user = await tx1.user.find_first_or_raise(where={'id': user.id})
297-
phantom_count = await tx1.user.count()
298-
299-
# No dirty read
300-
assert tx1_user.name == dirty_user.name
301-
# No non-repeatable read
302-
assert tx1_user.name == non_repeatable_user.name
303-
# Have phantom read
304-
assert tx1_count != phantom_count
305-
306-
307-
@pytest.mark.asyncio
308-
@pytest.mark.skipif(True, reason='Available for SQL Server only')
309-
async def test_snapshot_isolation_level() -> None:
310-
"""A transaction isolation level is set to `SNAPSHOT`"""
311-
raise NotImplementedError
312-
313-
314-
@pytest.mark.asyncio
315-
async def test_serializable_isolation_level(client: Prisma) -> None:
316-
"""A transaction isolation level is set to `SERIALIZABLE`"""
317-
client2 = Prisma()
318-
await client2.connect()
319-
320-
user = await client.user.create(data={'name': 'Robert'})
321-
322-
async with client.tx(isolation_level=prisma.TransactionIsolationLevel.SERIALIZABLE) as tx1:
323-
tx1_user = await tx1.user.find_first_or_raise(where={'id': user.id})
324-
tx1_count = await tx1.user.count()
325-
326-
async with client2.tx() as tx2:
327-
await tx2.user.update(data={'name': 'Tegan'}, where={'id': user.id})
328-
await tx2.user.create(data={'name': 'Bobby'})
218+
@pytest.mark.parametrize(
219+
('input_level', 'expected_level'),
220+
[
221+
pytest.param(
222+
prisma.TransactionIsolationLevel.READ_UNCOMMITTED,
223+
'READ_UNCOMMITTED',
224+
id='read uncommitted',
225+
marks=pytest.mark.skipif(CURRENT_DATABASE in ['cockroachdb', 'sqlite'], reason='Not available'),
226+
),
227+
pytest.param(
228+
prisma.TransactionIsolationLevel.READ_COMMITTED,
229+
'READ_COMMITTED',
230+
id='read committed',
231+
marks=pytest.mark.skipif(CURRENT_DATABASE in ['cockroachdb', 'sqlite'], reason='Not available'),
232+
),
233+
pytest.param(
234+
prisma.TransactionIsolationLevel.REPEATABLE_READ,
235+
'REPEATABLE_READ',
236+
id='repeatable read',
237+
marks=pytest.mark.skipif(CURRENT_DATABASE in ['cockroachdb', 'sqlite'], reason='Not available'),
238+
),
239+
pytest.param(
240+
prisma.TransactionIsolationLevel.SNAPSHOT,
241+
'SNAPSHOT',
242+
id='snapshot',
243+
marks=pytest.mark.skipif(True, reason='Available for SQL Server only'),
244+
),
245+
pytest.param(
246+
prisma.TransactionIsolationLevel.SERIALIZABLE,
247+
'SERIALIZABLE',
248+
id='serializable',
249+
marks=pytest.mark.skipif(
250+
CURRENT_DATABASE == 'sqlite', reason='PRAGMA has only effect in shared-cache mode'
251+
),
252+
),
253+
],
254+
)
255+
# TODO: remove after issue will be resolved
256+
@pytest.mark.skipif(CURRENT_DATABASE in ['mysql', 'mariadb'], reason='https://github.com/prisma/prisma/issues/22890')
257+
async def test_isolation_level(
258+
client: Prisma, raw_queries: RawQueries, input_level: prisma.TransactionIsolationLevel, expected_level: str
259+
) -> None:
260+
"""A transaction isolation level is set correctly"""
261+
async with client.tx(isolation_level=input_level) as tx:
262+
results = await tx.query_raw(raw_queries.select_tx_isolation)
329263

330-
dirty_user = await tx1.user.find_first_or_raise(where={'id': user.id})
264+
assert len(results) == 1
331265

332-
non_repeatable_user = await tx1.user.find_first_or_raise(where={'id': user.id})
333-
phantom_count = await tx1.user.count()
266+
row = results[0]
267+
assert any(row)
334268

335-
# No dirty read
336-
assert tx1_user.name == dirty_user.name
337-
# No non-repeatable read
338-
assert tx1_user.name == non_repeatable_user.name
339-
# No phantom read
340-
assert tx1_count == phantom_count
269+
level = next(iter(row.values()))
270+
# The result can depends on the database, so we do upper() and replace()
271+
level = str(level).upper().replace(' ', '_').replace('-', '_')
272+
assert level == expected_level

0 commit comments

Comments
 (0)