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
« prev ^ index » next coverage.py v7.2.7, created at 2024-08-27 18:25 +0000
1import pytest
3import prisma
4from prisma import Prisma
5from lib.testing import async_fixture
6from prisma.models import User, Types, Unique2
8from ..utils import CURRENT_DATABASE
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
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
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'
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
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
84 post_id = updated.posts[0].id
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 )
111 assert updated is not None
112 assert updated.posts is not None
113 assert len(updated.posts) == 0
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
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
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
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'
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
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
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 )
177 assert exc.match(r'published')
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'
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
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 )
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'