Coverage for databases/sync_tests/test_transactions.py: 100%

1import time 

2from typing import Optional 

3from datetime import timedelta 

4 

5import pytest 

6 

7import prisma 

8from prisma import Prisma 

9from prisma.models import User, Profile 

10 

11from ..utils import CURRENT_DATABASE 

12 

13 

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' 

19 

20 # ensure not commited outside transaction 

21 assert client.user.count() == 0 

22 

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 ) 

34 

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!' 

40 

41 

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' 

47 

48 # ensure not commited outside transaction 

49 assert client.user.count() == 0 

50 

51 transaction.profile.create( 

52 { 

53 'description': 'Hello, there!', 

54 'country': 'Scotland', 

55 'user': { 

56 'connect': { 

57 'id': user.id, 

58 }, 

59 }, 

60 }, 

61 ) 

62 

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!' 

68 

69 

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 

73 

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.') 

78 

79 assert exc.match('Error ocurred mid transaction.') 

80 

81 assert user is not None 

82 found = client.user.find_unique(where={'id': user.id}) 

83 assert found is None 

84 

85 

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'}) 

92 

93 assert client.user.count() == 0 

94 assert transaction.user.count() == 2 

95 

96 assert client.user.count() == 2 

97 

98 

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.1) 

106 

107 with pytest.raises(prisma.errors.TransactionExpiredError) as exc: 

108 transaction.user.create({'name': 'Robert'}) 

109 

110 raise exc.value 

111 

112 

113@pytest.mark.skipif(CURRENT_DATABASE == 'sqlite', reason='SQLite does not support concurrent writes') 

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'}) 

120 

121 assert tx1.user.find_first(where={'name': 'Robert'}) is None 

122 assert tx2.user.find_first(where={'name': 'Tegan'}) is None 

123 

124 found = tx1.user.find_first(where={'name': 'Tegan'}) 

125 assert found is not None 

126 assert found.id == user1.id 

127 

128 found = tx2.user.find_first(where={'name': 'Robert'}) 

129 assert found is not None 

130 assert found.id == user2.id 

131 

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 

136 

137 assert client.user.count() == 2 

138 

139 

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!') 

145 

146 

147@pytest.mark.skipif(CURRENT_DATABASE == 'sqlite', reason='SQLite does not support concurrent writes') 

148def test_transaction_within_transaction_warning(client: Prisma) -> None: 

149 """A warning is raised if a transaction is started from another transaction client""" 

150 tx1 = client.tx().start() 

151 with pytest.warns(UserWarning) as warnings: 

152 tx1.tx().start() 

153 

154 assert len(warnings) == 1 

155 record = warnings[0] 

156 assert not isinstance(record.message, str) 

157 assert ( 

158 record.message.args[0] 

159 == 'The current client is already in a transaction. This can lead to surprising behaviour.' 

160 ) 

161 assert record.filename == __file__ 

162 

163 

164@pytest.mark.skipif(CURRENT_DATABASE == 'sqlite', reason='SQLite does not support concurrent writes') 

165def test_transaction_within_transaction_context_warning( 

166 client: Prisma, 

167) -> None: 

168 """A warning is raised if a transaction is started from another transaction client""" 

169 with client.tx() as tx1: 

170 with pytest.warns(UserWarning) as warnings: 

171 with tx1.tx(): 

172 pass 

173 

174 assert len(warnings) == 1 

175 record = warnings[0] 

176 assert not isinstance(record.message, str) 

177 assert ( 

178 record.message.args[0] 

179 == 'The current client is already in a transaction. This can lead to surprising behaviour.' 

180 ) 

181 assert record.filename == __file__ 

182 

183 

184def test_transaction_not_started(client: Prisma) -> None: 

185 """A `TransactionNotStartedError` is raised when attempting to call `commit()` or `rollback()` 

186 on a transaction that hasn't been started yet. 

187 """ 

188 tx = client.tx() 

189 

190 with pytest.raises(prisma.errors.TransactionNotStartedError): 

191 tx.commit() 

192 

193 with pytest.raises(prisma.errors.TransactionNotStartedError): 

194 tx.rollback() 

195 

196 

197def test_transaction_already_closed(client: Prisma) -> None: 

198 """Attempting to use a transaction outside of the context block raises an error""" 

199 with client.tx() as transaction: 

200 pass 

201 

202 with pytest.raises(prisma.errors.TransactionExpiredError) as exc: 

203 transaction.user.delete_many() 

204 

205 assert exc.match('Transaction already closed')