1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
|
# type: ignore
import uuid
from typing import List, Optional
import ormar
import pydantic
import pytest
from asgi_lifespan import LifespanManager
from fastapi import FastAPI
from httpx import ASGITransport, AsyncClient
from tests.lifespan import init_tests, lifespan
from tests.settings import create_config
base_ormar_config = create_config()
app = FastAPI(lifespan=lifespan(base_ormar_config))
class Thing(ormar.Model):
ormar_config = base_ormar_config.copy(tablename="things")
id: uuid.UUID = ormar.UUID(primary_key=True, default=uuid.uuid4)
name: str = ormar.Text(default="")
js: pydantic.Json = ormar.JSON()
create_test_database = init_tests(base_ormar_config)
@app.get("/things", response_model=List[Thing])
async def read_things():
return await Thing.objects.order_by("name").all()
@app.get("/things_with_sample", response_model=List[Thing])
async def read_things_sample():
await Thing(name="b", js=["asdf", "asdf", "bobby", "nigel"]).save()
await Thing(name="a", js='["lemon", "raspberry", "lime", "pumice"]').save()
return await Thing.objects.order_by("name").all()
@app.get("/things_with_sample_after_init", response_model=Thing)
async def read_things_init():
thing1 = Thing(js="{}")
thing1.name = "d"
thing1.js = ["js", "set", "after", "constructor"]
await thing1.save()
return thing1
@app.put("/update_thing", response_model=Thing)
async def update_things(thing: Thing):
thing.js = ["js", "set", "after", "update"] # type: ignore
await thing.update()
return thing
@app.post("/things", response_model=Thing)
async def create_things(thing: Thing):
thing = await thing.save()
return thing
@app.get("/things_untyped")
async def read_things_untyped():
return await Thing.objects.order_by("name").all()
@pytest.mark.asyncio
async def test_json_is_required_if_not_nullable():
with pytest.raises(pydantic.ValidationError):
Thing()
@pytest.mark.asyncio
async def test_json_is_not_required_if_nullable():
class Thing2(ormar.Model):
ormar_config = base_ormar_config.copy(tablename="things2")
id: uuid.UUID = ormar.UUID(primary_key=True, default=uuid.uuid4)
name: str = ormar.Text(default="")
js: Optional[pydantic.Json] = ormar.JSON(nullable=True)
Thing2()
@pytest.mark.asyncio
async def test_setting_values_after_init():
async with base_ormar_config.database:
t1 = Thing(id="67a82813-d90c-45ff-b546-b4e38d7030d7", name="t1", js=["thing1"])
assert '["thing1"]' in t1.model_dump_json()
await t1.save()
t1.model_dump_json()
assert '["thing1"]' in t1.model_dump_json()
assert '["thing1"]' in (await Thing.objects.get(id=t1.id)).model_dump_json()
await t1.update()
assert '["thing1"]' in (await Thing.objects.get(id=t1.id)).model_dump_json()
@pytest.mark.asyncio
async def test_read_main():
transport = ASGITransport(app=app)
client = AsyncClient(transport=transport, base_url="http://testserver")
async with client as client, LifespanManager(app):
response = await client.get("/things_with_sample")
assert response.status_code == 200
# check if raw response not double encoded
assert '["lemon","raspberry","lime","pumice"]' in response.text
# parse json and check that we get lists not strings
resp = response.json()
assert resp[0].get("js") == ["lemon", "raspberry", "lime", "pumice"]
assert resp[1].get("js") == ["asdf", "asdf", "bobby", "nigel"]
# create a new one
response = await client.post(
"/things", json={"js": ["test", "test2"], "name": "c"}
)
assert response.json().get("js") == ["test", "test2"]
# get all with new one
response = await client.get("/things")
assert response.status_code == 200
assert '["test","test2"]' in response.text
resp = response.json()
assert resp[0].get("js") == ["lemon", "raspberry", "lime", "pumice"]
assert resp[1].get("js") == ["asdf", "asdf", "bobby", "nigel"]
assert resp[2].get("js") == ["test", "test2"]
response = await client.get("/things_with_sample_after_init")
assert response.status_code == 200
resp = response.json()
assert resp.get("js") == ["js", "set", "after", "constructor"]
# test new with after constructor
response = await client.get("/things")
resp = response.json()
assert resp[0].get("js") == ["lemon", "raspberry", "lime", "pumice"]
assert resp[1].get("js") == ["asdf", "asdf", "bobby", "nigel"]
assert resp[2].get("js") == ["test", "test2"]
assert resp[3].get("js") == ["js", "set", "after", "constructor"]
response = await client.put("/update_thing", json=resp[3])
assert response.status_code == 200
resp = response.json()
assert resp.get("js") == ["js", "set", "after", "update"]
# test new with after constructor
response = await client.get("/things_untyped")
resp = response.json()
assert resp[0].get("js") == ["lemon", "raspberry", "lime", "pumice"]
assert resp[1].get("js") == ["asdf", "asdf", "bobby", "nigel"]
assert resp[2].get("js") == ["test", "test2"]
assert resp[3].get("js") == ["js", "set", "after", "update"]
|
