Coverage for databases/tests/test_update.py: 100%

88 statements  

« prev     ^ index     » next       coverage.py v7.2.7, created at 2024-08-27 18:25 +0000

1import pytest 

2 

3import prisma 

4from prisma import Prisma 

5from lib.testing import async_fixture 

6from prisma.models import User, Types, Unique2 

7 

8from ..utils import CURRENT_DATABASE 

9 

10 

11@async_fixture(name='user_id') 

12async def user_id_fixture(client: Prisma) -> str: 

13 user = await client.user.create({'name': 'Robert'}) 

14 return user.id 

15 

16 

17@pytest.mark.asyncio 

18async def test_update(client: Prisma) -> None: 

19 """Standard usage""" 

20 post = await client.post.create( 

21 { 

22 'title': 'Hi from Create!', 

23 'published': True, 

24 'description': 'Prisma is a database toolkit that makes databases easy.', 

25 'author': { 

26 'create': { 

27 'name': 'Bob', 

28 }, 

29 }, 

30 } 

31 ) 

32 assert post.author is None 

33 assert post.title == 'Hi from Create!' 

34 updated = await client.post.update( 

35 where={ 

36 'id': post.id, 

37 }, 

38 data={ 

39 'title': 'Hi from Update!', 

40 }, 

41 ) 

42 assert updated is not None 

43 assert updated.title == 'Hi from Update!' 

44 assert updated.updated_at != post.updated_at 

45 assert updated.created_at == post.created_at 

46 

47 updated = await client.post.update( 

48 where={ 

49 'id': post.id, 

50 }, 

51 include={ 

52 'author': True, 

53 }, 

54 data={ 

55 'published': False, 

56 'description': 'Updated desc.', 

57 }, 

58 ) 

59 assert updated is not None 

60 assert updated.published is False 

61 assert updated.description == 'Updated desc.' 

62 assert updated.author is not None 

63 assert updated.author.name == 'Bob' 

64 

65 

66@pytest.mark.asyncio 

67@pytest.mark.parametrize('method', ['disconnect', 'delete']) 

68async def test_update_with_create_disconnect(client: Prisma, user_id: str, method: str) -> None: 

69 """Removing a relational field""" 

70 user = await client.user.find_unique(where={'id': user_id}, include={'posts': True}) 

71 assert user is not None 

72 assert user.posts is not None 

73 assert len(user.posts) == 0 

74 

75 updated = await client.user.update( 

76 where={'id': user_id}, 

77 data={'posts': {'create': [{'title': 'My post', 'published': True}]}}, 

78 include={'posts': True}, 

79 ) 

80 assert updated is not None 

81 assert updated.posts is not None 

82 assert len(updated.posts) == 1 

83 

84 post_id = updated.posts[0].id 

85 

86 if method == 'disconnect': 

87 updated = await client.user.update( 

88 where={'id': user_id}, 

89 data={ 

90 'posts': { 

91 'disconnect': [ 

92 {'id': post_id}, 

93 ] 

94 } 

95 }, 

96 include={'posts': True}, 

97 ) 

98 else: 

99 updated = await client.user.update( 

100 where={'id': user_id}, 

101 data={ 

102 'posts': { 

103 'delete': [ 

104 {'id': post_id}, 

105 ], 

106 } 

107 }, 

108 include={'posts': True}, 

109 ) 

110 

111 assert updated is not None 

112 assert updated.posts is not None 

113 assert len(updated.posts) == 0 

114 

115 

116@pytest.mark.asyncio 

117async def test_atomic_update(client: Prisma) -> None: 

118 """Atomically incrementing a value by 1""" 

119 post = await client.post.create({'title': 'My Post', 'published': False}) 

120 assert post.title == 'My Post' 

121 assert post.views == 0 

122 

123 updated = await client.post.update(where={'id': post.id}, data={'views': {'increment': 1}}) 

124 assert updated is not None 

125 assert updated.views == 1 

126 

127 

128@pytest.mark.asyncio 

129async def test_update_record_not_found(client: Prisma) -> None: 

130 """Updating a non-existent record returns None""" 

131 post = await client.post.update(where={'id': 'wow'}, data={'title': 'Hi from Update!'}) 

132 assert post is None 

133 

134 

135@pytest.mark.asyncio 

136async def test_setting_field_to_null(client: Prisma) -> None: 

137 """Updating a field to None sets the database record to None""" 

138 post = await client.post.create( 

139 data={ 

140 'title': 'Post', 

141 'published': False, 

142 'description': 'My description', 

143 }, 

144 ) 

145 assert post.description == 'My description' 

146 

147 updated = await client.post.update( 

148 where={ 

149 'id': post.id, 

150 }, 

151 data={'description': None}, 

152 ) 

153 assert updated is not None 

154 assert updated.id == post.id 

155 assert updated.description is None 

156 

157 

158@pytest.mark.asyncio 

159async def test_setting_non_nullable_field_to_null(client: Prisma) -> None: 

160 """Attempting to set a non-nullable field to null raises an error""" 

161 post = await client.post.create( 

162 data={ 

163 'title': 'Post', 

164 'published': False, 

165 }, 

166 ) 

167 assert post.published is False 

168 

169 with pytest.raises(prisma.errors.MissingRequiredValueError) as exc: 

170 await client.post.update( 

171 where={ 

172 'id': post.id, 

173 }, 

174 data={'published': None}, # type: ignore 

175 ) 

176 

177 assert exc.match(r'published') 

178 

179 

180@pytest.mark.prisma 

181@pytest.mark.asyncio 

182async def test_update_id_field() -> None: 

183 """Setting an ID field""" 

184 user = await User.prisma().create( 

185 data={ 

186 'name': 'Robert', 

187 }, 

188 ) 

189 updated = await User.prisma().update( 

190 where={ 

191 'id': user.id, 

192 }, 

193 data={ 

194 'id': 'abcd123', 

195 }, 

196 ) 

197 assert updated is not None 

198 assert updated.id == 'abcd123' 

199 

200 

201@pytest.mark.prisma 

202@pytest.mark.asyncio 

203@pytest.mark.skipif( 

204 CURRENT_DATABASE == 'cockroachdb', 

205 reason='https://github.com/prisma/prisma/issues/16612', 

206) 

207async def test_update_id_field_atomic() -> None: 

208 """Setting an ID field atomically""" 

209 record = await Types.prisma().create({}) 

210 updated = await Types.prisma().update( 

211 where={ 

212 'id': record.id, 

213 }, 

214 data={ 

215 'id': { 

216 'increment': 500, 

217 }, 

218 }, 

219 ) 

220 assert updated is not None 

221 assert updated.id == record.id + 500 

222 

223 

224@pytest.mark.prisma 

225@pytest.mark.asyncio 

226async def test_update_unique_field() -> None: 

227 """Setting a unique field""" 

228 record = await Unique2.prisma().create( 

229 data={ 

230 'name': 'Robert', 

231 'surname': 'Craigie', 

232 } 

233 ) 

234 

235 updated = await Unique2.prisma().update( 

236 where={ 

237 'surname': record.surname, 

238 }, 

239 data={ 

240 'surname': 'George', 

241 }, 

242 ) 

243 assert updated is not None 

244 assert updated.name == record.name 

245 assert updated.surname == 'George'