Coverage for databases/sync_tests/test_transactions.py: 100%
90 statements
« prev ^ index » next coverage.py v7.2.7, created at 2024-04-28 15:17 +0000
« prev ^ index » next coverage.py v7.2.7, created at 2024-04-28 15:17 +0000
1import time
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
14def test_model_query(client: Prisma) -> None:
15 """Basic usage within model queries"""
16 with client.tx(timeout=timedelta(milliseconds=1000)) as tx:
17 user = User.prisma(tx).create({'name': 'Robert'})
18 assert user.name == 'Robert'
20 # ensure not commited outside transaction
21 assert client.user.count() == 0
23 Profile.prisma(tx).create(
24 {
25 'description': 'Hello, there!',
26 'country': 'Scotland',
27 'user': {
28 'connect': {
29 'id': user.id,
30 },
31 },
32 },
33 )
35 found = client.user.find_unique(where={'id': user.id}, include={'profile': True})
36 assert found is not None
37 assert found.name == 'Robert'
38 assert found.profile is not None
39 assert found.profile.description == 'Hello, there!'
42def test_context_manager(client: Prisma) -> None:
43 """Basic usage within a context manager"""
44 with client.tx(timeout=timedelta(milliseconds=1000)) as transaction:
45 user = transaction.user.create({'name': 'Robert'})
46 assert user.name == 'Robert'
48 # ensure not commited outside transaction
49 assert client.user.count() == 0
51 transaction.profile.create(
52 {
53 'description': 'Hello, there!',
54 'country': 'Scotland',
55 'user': {
56 'connect': {
57 'id': user.id,
58 },
59 },
60 },
61 )
63 found = client.user.find_unique(where={'id': user.id}, include={'profile': True})
64 assert found is not None
65 assert found.name == 'Robert'
66 assert found.profile is not None
67 assert found.profile.description == 'Hello, there!'
70def test_context_manager_auto_rollback(client: Prisma) -> None:
71 """An error being thrown when within a context manager causes the transaction to be rolled back"""
72 user: Optional[User] = None
74 with pytest.raises(RuntimeError) as exc:
75 with client.tx() as tx:
76 user = tx.user.create({'name': 'Tegan'})
77 raise RuntimeError('Error ocurred mid transaction.')
79 assert exc.match('Error ocurred mid transaction.')
81 assert user is not None
82 found = client.user.find_unique(where={'id': user.id})
83 assert found is None
86def test_batch_within_transaction(client: Prisma) -> None:
87 """Query batching can be used within transactions"""
88 with client.tx(timeout=timedelta(milliseconds=10000)) as transaction:
89 with transaction.batch_() as batcher:
90 batcher.user.create({'name': 'Tegan'})
91 batcher.user.create({'name': 'Robert'})
93 assert client.user.count() == 0
94 assert transaction.user.count() == 2
96 assert client.user.count() == 2
99def test_timeout(client: Prisma) -> None:
100 """A `TransactionExpiredError` is raised when the transaction times out."""
101 # this outer block is necessary becuse to the context manager it appears that no error
102 # ocurred so it will attempt to commit the transaction, triggering the expired error again
103 with pytest.raises(prisma.errors.TransactionExpiredError):
104 with client.tx(timeout=timedelta(milliseconds=50)) as transaction:
105 time.sleep(0.05)
107 with pytest.raises(prisma.errors.TransactionExpiredError) as exc:
108 transaction.user.create({'name': 'Robert'})
110 raise exc.value
113@pytest.mark.skipif(CURRENT_DATABASE == 'sqlite', reason='This is currently broken...')
114def test_concurrent_transactions(client: Prisma) -> None:
115 """Two separate transactions can be used independently of each other at the same time"""
116 timeout = timedelta(milliseconds=15000)
117 with client.tx(timeout=timeout) as tx1, client.tx(timeout=timeout) as tx2:
118 user1 = tx1.user.create({'name': 'Tegan'})
119 user2 = tx2.user.create({'name': 'Robert'})
121 assert tx1.user.find_first(where={'name': 'Robert'}) is None
122 assert tx2.user.find_first(where={'name': 'Tegan'}) is None
124 found = tx1.user.find_first(where={'name': 'Tegan'})
125 assert found is not None
126 assert found.id == user1.id
128 found = tx2.user.find_first(where={'name': 'Robert'})
129 assert found is not None
130 assert found.id == user2.id
132 # ensure not leaked
133 assert client.user.count() == 0
134 assert (tx1.user.find_first(where={'name': user2.name})) is None
135 assert (tx2.user.find_first(where={'name': user1.name})) is None
137 assert client.user.count() == 2
140def test_transaction_raises_original_error(client: Prisma) -> None:
141 """If an error is raised during the execution of the transaction, it is raised"""
142 with pytest.raises(RuntimeError, match=r'Test error!'):
143 with client.tx():
144 raise RuntimeError('Test error!')
147def test_transaction_within_transaction_warning(client: Prisma) -> None:
148 """A warning is raised if a transaction is started from another transaction client"""
149 tx1 = client.tx().start()
150 with pytest.warns(UserWarning) as warnings:
151 tx1.tx().start()
153 assert len(warnings) == 1
154 record = warnings[0]
155 assert not isinstance(record.message, str)
156 assert (
157 record.message.args[0]
158 == 'The current client is already in a transaction. This can lead to surprising behaviour.'
159 )
160 assert record.filename == __file__
163def test_transaction_within_transaction_context_warning(
164 client: Prisma,
165) -> None:
166 """A warning is raised if a transaction is started from another transaction client"""
167 with client.tx() as tx1:
168 with pytest.warns(UserWarning) as warnings:
169 with tx1.tx():
170 pass
172 assert len(warnings) == 1
173 record = warnings[0]
174 assert not isinstance(record.message, str)
175 assert (
176 record.message.args[0]
177 == 'The current client is already in a transaction. This can lead to surprising behaviour.'
178 )
179 assert record.filename == __file__
182def test_transaction_not_started(client: Prisma) -> None:
183 """A `TransactionNotStartedError` is raised when attempting to call `commit()` or `rollback()`
184 on a transaction that hasn't been started yet.
185 """
186 tx = client.tx()
188 with pytest.raises(prisma.errors.TransactionNotStartedError):
189 tx.commit()
191 with pytest.raises(prisma.errors.TransactionNotStartedError):
192 tx.rollback()
195def test_transaction_already_closed(client: Prisma) -> None:
196 """Attempting to use a transaction outside of the context block raises an error"""
197 with client.tx() as transaction:
198 pass
200 with pytest.raises(prisma.errors.TransactionExpiredError) as exc:
201 transaction.user.delete_many()
203 assert exc.match('Transaction already closed')