Coverage for databases/tests/test_transactions.py: 100%
101 statements
« prev ^ index » next coverage.py v7.2.7, created at 2024-04-28 15:17 +0000
1import asyncio
2from typing import Optional
3from datetime import timedelta
5import pytest
7import prisma
8from prisma import Prisma
9from prisma.models import User, Profile
11from ..utils import CURRENT_DATABASE
14@pytest.mark.asyncio
15async def test_model_query(client: Prisma) -> None:
16 """Basic usage within model queries"""
17 async with client.tx(timeout=timedelta(milliseconds=1000)) as tx:
18 user = await User.prisma(tx).create({'name': 'Robert'})
19 assert user.name == 'Robert'
21 # ensure not commited outside transaction
22 assert await client.user.count() == 0
24 await Profile.prisma(tx).create(
25 {
26 'description': 'Hello, there!',
27 'country': 'Scotland',
28 'user': {
29 'connect': {
30 'id': user.id,
31 },
32 },
33 },
34 )
36 found = await client.user.find_unique(where={'id': user.id}, include={'profile': True})
37 assert found is not None
38 assert found.name == 'Robert'
39 assert found.profile is not None
40 assert found.profile.description == 'Hello, there!'
43@pytest.mark.asyncio
44async def test_context_manager(client: Prisma) -> None:
45 """Basic usage within a context manager"""
46 async with client.tx(timeout=timedelta(milliseconds=1000)) as transaction:
47 user = await transaction.user.create({'name': 'Robert'})
48 assert user.name == 'Robert'
50 # ensure not commited outside transaction
51 assert await client.user.count() == 0
53 await transaction.profile.create(
54 {
55 'description': 'Hello, there!',
56 'country': 'Scotland',
57 'user': {
58 'connect': {
59 'id': user.id,
60 },
61 },
62 },
63 )
65 found = await client.user.find_unique(where={'id': user.id}, include={'profile': True})
66 assert found is not None
67 assert found.name == 'Robert'
68 assert found.profile is not None
69 assert found.profile.description == 'Hello, there!'
72@pytest.mark.asyncio
73async def test_context_manager_auto_rollback(client: Prisma) -> None:
74 """An error being thrown when within a context manager causes the transaction to be rolled back"""
75 user: Optional[User] = None
77 with pytest.raises(RuntimeError) as exc:
78 async with client.tx() as tx:
79 user = await tx.user.create({'name': 'Tegan'})
80 raise RuntimeError('Error ocurred mid transaction.')
82 assert exc.match('Error ocurred mid transaction.')
84 assert user is not None
85 found = await client.user.find_unique(where={'id': user.id})
86 assert found is None
89@pytest.mark.asyncio
90async def test_batch_within_transaction(client: Prisma) -> None:
91 """Query batching can be used within transactions"""
92 async with client.tx(timeout=timedelta(milliseconds=10000)) as transaction:
93 async with transaction.batch_() as batcher:
94 batcher.user.create({'name': 'Tegan'})
95 batcher.user.create({'name': 'Robert'})
97 assert await client.user.count() == 0
98 assert await transaction.user.count() == 2
100 assert await client.user.count() == 2
103@pytest.mark.asyncio
104async def test_timeout(client: Prisma) -> None:
105 """A `TransactionExpiredError` is raised when the transaction times out."""
106 # this outer block is necessary becuse to the context manager it appears that no error
107 # ocurred so it will attempt to commit the transaction, triggering the expired error again
108 with pytest.raises(prisma.errors.TransactionExpiredError):
109 async with client.tx(timeout=timedelta(milliseconds=50)) as transaction:
110 await asyncio.sleep(0.05)
112 with pytest.raises(prisma.errors.TransactionExpiredError) as exc:
113 await transaction.user.create({'name': 'Robert'})
115 raise exc.value
118@pytest.mark.asyncio
119@pytest.mark.skipif(CURRENT_DATABASE == 'sqlite', reason='This is currently broken...')
120async def test_concurrent_transactions(client: Prisma) -> None:
121 """Two separate transactions can be used independently of each other at the same time"""
122 timeout = timedelta(milliseconds=15000)
123 async with client.tx(timeout=timeout) as tx1, client.tx(timeout=timeout) as tx2:
124 user1 = await tx1.user.create({'name': 'Tegan'})
125 user2 = await tx2.user.create({'name': 'Robert'})
127 assert await tx1.user.find_first(where={'name': 'Robert'}) is None
128 assert await tx2.user.find_first(where={'name': 'Tegan'}) is None
130 found = await tx1.user.find_first(where={'name': 'Tegan'})
131 assert found is not None
132 assert found.id == user1.id
134 found = await tx2.user.find_first(where={'name': 'Robert'})
135 assert found is not None
136 assert found.id == user2.id
138 # ensure not leaked
139 assert await client.user.count() == 0
140 assert (await tx1.user.find_first(where={'name': user2.name})) is None
141 assert (await tx2.user.find_first(where={'name': user1.name})) is None
143 assert await client.user.count() == 2
146@pytest.mark.asyncio
147async def test_transaction_raises_original_error(client: Prisma) -> None:
148 """If an error is raised during the execution of the transaction, it is raised"""
149 with pytest.raises(RuntimeError, match=r'Test error!'):
150 async with client.tx():
151 raise RuntimeError('Test error!')
154@pytest.mark.asyncio
155async def test_transaction_within_transaction_warning(client: Prisma) -> None:
156 """A warning is raised if a transaction is started from another transaction client"""
157 tx1 = await client.tx().start()
158 with pytest.warns(UserWarning) as warnings:
159 await tx1.tx().start()
161 assert len(warnings) == 1
162 record = warnings[0]
163 assert not isinstance(record.message, str)
164 assert (
165 record.message.args[0]
166 == 'The current client is already in a transaction. This can lead to surprising behaviour.'
167 )
168 assert record.filename == __file__
171@pytest.mark.asyncio
172async def test_transaction_within_transaction_context_warning(
173 client: Prisma,
174) -> None:
175 """A warning is raised if a transaction is started from another transaction client"""
176 async with client.tx() as tx1:
177 with pytest.warns(UserWarning) as warnings:
178 async with tx1.tx():
179 pass
181 assert len(warnings) == 1
182 record = warnings[0]
183 assert not isinstance(record.message, str)
184 assert (
185 record.message.args[0]
186 == 'The current client is already in a transaction. This can lead to surprising behaviour.'
187 )
188 assert record.filename == __file__
191@pytest.mark.asyncio
192async def test_transaction_not_started(client: Prisma) -> None:
193 """A `TransactionNotStartedError` is raised when attempting to call `commit()` or `rollback()`
194 on a transaction that hasn't been started yet.
195 """
196 tx = client.tx()
198 with pytest.raises(prisma.errors.TransactionNotStartedError):
199 await tx.commit()
201 with pytest.raises(prisma.errors.TransactionNotStartedError):
202 await tx.rollback()
205@pytest.mark.asyncio
206async def test_transaction_already_closed(client: Prisma) -> None:
207 """Attempting to use a transaction outside of the context block raises an error"""
208 async with client.tx() as transaction:
209 pass
211 with pytest.raises(prisma.errors.TransactionExpiredError) as exc:
212 await transaction.user.delete_many()
214 assert exc.match('Transaction already closed')