Coverage for tests/test_builder.py: 100%
98 statements
« prev ^ index » next coverage.py v7.2.7, created at 2024-08-27 18:25 +0000
1from __future__ import annotations
3import datetime
4from typing import Any, Optional
6import pytest
7from pydantic import BaseModel
8from inline_snapshot import snapshot
10from prisma import PrismaMethod, models
11from prisma.bases import _PrismaModel as PrismaModel
12from prisma.utils import _NoneType
13from prisma.errors import UnknownModelError, UnknownRelationalFieldError
14from prisma._compat import PYDANTIC_V2
15from prisma._builder import QueryBuilder, serializer
16from prisma.metadata import PRISMA_MODELS, RELATIONAL_FIELD_MAPPINGS
18# TODO: more tests
19# TODO: cleanup registered serializers
20# TODO: these tests should be schema agnostic
21# one schema should be used for all these tests
22# and only changed when needed for these tests
23# otherwise, simply adding a field to a model
24# will break these tests
27def build_query(
28 method: PrismaMethod,
29 arguments: dict[str, Any],
30 **kwargs: Any,
31) -> str:
32 return QueryBuilder(
33 method=method,
34 arguments=arguments,
35 **kwargs,
36 prisma_models=PRISMA_MODELS,
37 relational_field_mappings=RELATIONAL_FIELD_MAPPINGS,
38 ).build_query()
41def test_basic_building() -> None:
42 """Standard builder usage with and without a model"""
43 query = QueryBuilder(
44 method='find_unique',
45 model=models.User,
46 arguments={'where': {'id': '1'}},
47 prisma_models=PRISMA_MODELS,
48 relational_field_mappings=RELATIONAL_FIELD_MAPPINGS,
49 ).build_query()
50 assert query == snapshot(
51 """\
52query {
53 result: findUniqueUser
54 (
55 where: {
56 id: "1"
57 }
58 )
59 {
60 id
61 name
62 email
63 created_at
64 }
65}\
66"""
67 )
69 query = QueryBuilder(
70 method='query_raw',
71 arguments={'where': {'id': '1'}},
72 prisma_models=PRISMA_MODELS,
73 relational_field_mappings=RELATIONAL_FIELD_MAPPINGS,
74 ).build_query()
75 assert query == snapshot(
76 """\
77mutation {
78 result: queryRaw
79 (
80 where: {
81 id: "1"
82 }
83 )
84}\
85"""
86 )
89def test_invalid_include() -> None:
90 """Invalid include field raises error"""
91 with pytest.raises(UnknownRelationalFieldError) as exception:
92 QueryBuilder(
93 method='find_unique',
94 model=models.User,
95 arguments={
96 'include': {
97 'hello': True,
98 }
99 },
100 prisma_models=PRISMA_MODELS,
101 relational_field_mappings=RELATIONAL_FIELD_MAPPINGS,
102 ).build_query()
104 assert exception.match('Field: "hello" either does not exist or is not a relational field on the User model')
107def test_include_no_model() -> None:
108 """Trying to include a field without acess to a model raises an error"""
109 with pytest.raises(ValueError) as exc:
110 build_query(
111 method='query_raw',
112 arguments={'include': {'posts': True}},
113 )
115 assert exc.match('Cannot include fields when model is None.')
118def test_include_with_arguments() -> None:
119 """Including a field with filters"""
120 query = QueryBuilder(
121 method='find_unique',
122 model=models.User,
123 arguments={'where': {'id': 1}, 'include': {'posts': {'where': {'id': 1}}}},
124 prisma_models=PRISMA_MODELS,
125 relational_field_mappings=RELATIONAL_FIELD_MAPPINGS,
126 ).build_query()
127 assert query == snapshot(
128 """\
129query {
130 result: findUniqueUser
131 (
132 where: {
133 id: 1
134 }
135 )
136 {
137 id
138 name
139 email
140 created_at
141 posts(
142 where: {
143 id: 1
144 }
145 )
146 {
147 id
148 created_at
149 updated_at
150 title
151 published
152 views
153 desc
154 author_id
155 }
156 }
157}\
158"""
159 )
162def test_raw_queries() -> None:
163 """Raw queries serialise paramaters to JSON"""
164 query = QueryBuilder(
165 method='query_raw',
166 arguments={
167 'query': 'SELECT * FROM User where id = $1',
168 'parameters': ['1263526'],
169 },
170 prisma_models=PRISMA_MODELS,
171 relational_field_mappings=RELATIONAL_FIELD_MAPPINGS,
172 ).build_query()
173 assert query == snapshot(
174 """\
175mutation {
176 result: queryRaw
177 (
178 query: "SELECT * FROM User where id = $1"
179 parameters: "[\\"1263526\\"]"
180 )
181}\
182"""
183 )
186def test_datetime_serialization_tz_aware() -> None:
187 """Serializing a timezone aware datetime converts to UTC"""
188 query = QueryBuilder(
189 method='find_unique',
190 model=models.Post,
191 arguments={'where': {'created_at': datetime.datetime(1985, 10, 26, 1, 1, 1, tzinfo=datetime.timezone.max)}},
192 prisma_models=PRISMA_MODELS,
193 relational_field_mappings=RELATIONAL_FIELD_MAPPINGS,
194 ).build_query()
195 assert query == snapshot(
196 """\
197query {
198 result: findUniquePost
199 (
200 where: {
201 created_at: "1985-10-25T01:02:01+00:00"
202 }
203 )
204 {
205 id
206 created_at
207 updated_at
208 title
209 published
210 views
211 desc
212 author_id
213 }
214}\
215"""
216 )
219def test_datetime_serialization_tz_unaware() -> None:
220 """Serializing a timezone naive datetime converts to UTC"""
221 query = QueryBuilder(
222 method='find_unique',
223 model=models.Post,
224 arguments={
225 'where': {
226 'created_at': datetime.datetime(1985, 10, 26, 1, 1, 1),
227 }
228 },
229 prisma_models=PRISMA_MODELS,
230 relational_field_mappings=RELATIONAL_FIELD_MAPPINGS,
231 ).build_query()
232 assert query == snapshot(
233 """\
234query {
235 result: findUniquePost
236 (
237 where: {
238 created_at: "1985-10-26T01:01:01+00:00"
239 }
240 )
241 {
242 id
243 created_at
244 updated_at
245 title
246 published
247 views
248 desc
249 author_id
250 }
251}\
252"""
253 )
256def test_unicode() -> None:
257 """Serializing unicode strings does not convert to ASCII"""
258 query = QueryBuilder(
259 method='find_unique',
260 model=models.User,
261 arguments={
262 'where': {
263 'name': '❤',
264 }
265 },
266 prisma_models=PRISMA_MODELS,
267 relational_field_mappings=RELATIONAL_FIELD_MAPPINGS,
268 ).build_query()
269 assert query == snapshot(
270 """\
271query {
272 result: findUniqueUser
273 (
274 where: {
275 name: "❤"
276 }
277 )
278 {
279 id
280 name
281 email
282 created_at
283 }
284}\
285"""
286 )
289def test_unknown_model() -> None:
290 """Passing unknown model raises an error"""
292 class FooModel(PrismaModel):
293 __prisma_model__ = 'Foo'
295 with pytest.raises(UnknownModelError) as exc:
296 QueryBuilder(
297 method='find_unique',
298 model=FooModel,
299 arguments={},
300 prisma_models=PRISMA_MODELS,
301 relational_field_mappings=RELATIONAL_FIELD_MAPPINGS,
302 ).build_query()
304 assert exc.match(r'Model: "Foo" does not exist\.')
307def test_unserializable_type() -> None:
308 """Passing an unserializable type raises an error"""
309 with pytest.raises(TypeError) as exc:
310 QueryBuilder(
311 method='find_first',
312 arguments={'where': QueryBuilder},
313 prisma_models=PRISMA_MODELS,
314 relational_field_mappings=RELATIONAL_FIELD_MAPPINGS,
315 ).build_query()
317 assert exc.match(r'Type <class \'prisma._builder.QueryBuilder\'> not serializable')
320def test_unserializable_instance() -> None:
321 """Passing an unserializable instance raises an error"""
322 with pytest.raises(TypeError) as exc:
323 QueryBuilder(
324 method='find_first',
325 arguments={
326 'where': _NoneType(),
327 },
328 prisma_models=PRISMA_MODELS,
329 relational_field_mappings=RELATIONAL_FIELD_MAPPINGS,
330 ).build_query()
332 assert exc.match(r'Type <class \'prisma.utils._NoneType\'> not serializable')
335def test_custom_serialization() -> None:
336 """Registering a custom serializer serializes as expected"""
338 class Foo:
339 def __init__(self, arg: int) -> None:
340 self.arg = arg
342 @serializer.register(Foo)
343 def custom_serializer(inst: Foo) -> int: # pyright: ignore[reportUnusedFunction]
344 return inst.arg
346 query = QueryBuilder(
347 method='find_unique',
348 model=models.Post,
349 arguments={
350 'where': {
351 'title': Foo(1),
352 }
353 },
354 prisma_models=PRISMA_MODELS,
355 relational_field_mappings=RELATIONAL_FIELD_MAPPINGS,
356 ).build_query()
357 assert query == snapshot(
358 """\
359query {
360 result: findUniquePost
361 (
362 where: {
363 title: 1
364 }
365 )
366 {
367 id
368 created_at
369 updated_at
370 title
371 published
372 views
373 desc
374 author_id
375 }
376}\
377"""
378 )
381def test_select() -> None:
382 """Selecting a subset of fields"""
384 class OtherModel(PrismaModel):
385 name: str
386 __prisma_model__ = 'User'
388 class CustomModel(PrismaModel):
389 published: bool
390 author: Optional[OtherModel]
392 __prisma_model__ = 'Post'
394 if not PYDANTIC_V2:
395 CustomModel.update_forward_refs(**locals()) # pyright: ignore[reportDeprecated]
397 query = QueryBuilder(
398 method='find_first',
399 model=CustomModel,
400 arguments={
401 'where': {
402 'title': 'Foo',
403 },
404 },
405 prisma_models=PRISMA_MODELS,
406 relational_field_mappings=RELATIONAL_FIELD_MAPPINGS,
407 ).build_query()
408 assert query == snapshot(
409 """\
410query {
411 result: findFirstPost
412 (
413 where: {
414 title: "Foo"
415 }
416 )
417 {
418 published
419 }
420}\
421"""
422 )
424 with pytest.raises(UnknownRelationalFieldError) as exc:
425 QueryBuilder(
426 method='find_unique',
427 model=OtherModel,
428 arguments={
429 'include': {
430 'posts': True,
431 },
432 },
433 prisma_models=PRISMA_MODELS,
434 relational_field_mappings=RELATIONAL_FIELD_MAPPINGS,
435 ).build_query()
437 assert exc.match(r'Field: "posts" either does not exist or is not a relational field on the OtherModel model')
439 query = QueryBuilder(
440 method='find_first',
441 model=CustomModel,
442 arguments={
443 'include': {
444 'author': True,
445 },
446 },
447 prisma_models=PRISMA_MODELS,
448 relational_field_mappings=RELATIONAL_FIELD_MAPPINGS,
449 ).build_query()
450 assert query == snapshot(
451 """\
452query {
453 result: findFirstPost
454 {
455 published
456 author {
457 name
458 }
459 }
460}\
461"""
462 )
465def test_select_non_prisma_model_basemodel() -> None:
466 """Fields that point to a `BaseModel` but do not set the `__prisma_model__`
467 class variable are included by default as scalar fields.
468 """
470 class OtherModel(PrismaModel):
471 name: str
472 __prisma_model__ = 'User'
474 class TypedJson(BaseModel):
475 foo: str
477 class CustomModel(PrismaModel):
478 published: bool
479 my_json_blob: TypedJson
480 author: Optional[OtherModel]
482 __prisma_model__ = 'Post'
484 if not PYDANTIC_V2:
485 CustomModel.update_forward_refs(**locals()) # pyright: ignore[reportDeprecated]
487 query = QueryBuilder(
488 method='find_first',
489 model=CustomModel,
490 arguments={
491 'where': {
492 'title': 'Foo',
493 },
494 },
495 prisma_models=PRISMA_MODELS,
496 relational_field_mappings=RELATIONAL_FIELD_MAPPINGS,
497 ).build_query()
498 assert query == snapshot(
499 """\
500query {
501 result: findFirstPost
502 (
503 where: {
504 title: "Foo"
505 }
506 )
507 {
508 published
509 my_json_blob
510 }
511}\
512"""
513 )
516@pytest.mark.skipif(PYDANTIC_V2, reason='Pydantic v2 has better forward refs support')
517def test_forward_reference_error() -> None:
518 """If a Model has forward references and doesn't call `update_forward_refs()` then we can't
519 tell if a given field should be included by default. Hence the sanity check here.
520 """
522 class TypedJson(BaseModel):
523 foo: str
525 class CustomModel(PrismaModel):
526 meta: TypedJson
528 __prisma_model__ = 'Post'
530 with pytest.raises(
531 RuntimeError, match=r'Forward references must be evaluated using CustomModel\.update_forward_refs\(\)'
532 ):
533 QueryBuilder(
534 method='find_first',
535 model=CustomModel,
536 arguments={},
537 prisma_models=PRISMA_MODELS,
538 relational_field_mappings=RELATIONAL_FIELD_MAPPINGS,
539 ).build_query()