-
Notifications
You must be signed in to change notification settings - Fork 15
Expand file tree
/
Copy pathtest_update_objects.py
More file actions
188 lines (172 loc) · 5.64 KB
/
test_update_objects.py
File metadata and controls
188 lines (172 loc) · 5.64 KB
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
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
import pytest
from httpx import AsyncClient
from sqlalchemy.ext.asyncio import AsyncSession
from starlette import status
from tests.misc.utils import fake
from tests.models import Computer, User, UserBio
from tests.schemas import UserAttributesBaseSchema, UserBioAttributesBaseSchema
pytestmark = pytest.mark.asyncio
class TestAtomicUpdateObjects:
async def test_update_two_objects(
self,
client: AsyncClient,
async_session: AsyncSession,
user_1: User,
user_1_bio: UserBio,
):
user_data = UserAttributesBaseSchema.from_orm(user_1)
user_bio_data = UserBioAttributesBaseSchema.from_orm(user_1_bio)
user_data.name = fake.name()
user_bio_data.favourite_movies = fake.sentence()
assert user_1.name != user_data.name
assert user_1_bio.favourite_movies != user_bio_data.favourite_movies
data_atomic_request = {
"atomic:operations": [
{
"op": "update",
"data": {
"id": str(user_1.id),
"type": "user",
"attributes": user_data.dict(),
},
},
{
"op": "update",
"data": {
"id": str(user_1_bio.id),
"type": "user_bio",
"attributes": user_bio_data.dict(),
},
},
],
}
response = await client.post("/operations", json=data_atomic_request)
assert response.status_code == status.HTTP_200_OK, response.text
response_data = response.json()
assert "atomic:results" in response_data, response_data
results = response_data["atomic:results"]
assert results
await async_session.refresh(user_1)
await async_session.refresh(user_1_bio)
assert user_1.name == user_data.name
assert user_1_bio.favourite_movies == user_bio_data.favourite_movies
assert results == [
{
"data": {
"attributes": user_data.dict(),
"id": str(user_1.id),
"type": "user",
},
"meta": None,
},
{
"data": {
"attributes": user_bio_data.dict(),
"id": str(user_1_bio.id),
"type": "user_bio",
},
"meta": None,
},
]
@pytest.mark.skip("todo: create relationships resources")
async def test_update_to_one_relationship_atomic(
self,
client: AsyncClient,
async_session: AsyncSession,
user_1: User,
computer_1: Computer,
):
"""
# as in doc
https://jsonapi.org/ext/atomic/#auto-id-updating-to-one-relationships
{
"atomic:operations": [{
"op": "update",
"ref": {
"type": "articles",
"id": "13",
"relationship": "author"
},
"data": {
"type": "people",
"id": "9"
}
}]
}
:param client:
:param async_session:
:param user_1:
:param computer_1:
:return:
"""
assert computer_1.user_id is None
operation_data = {
"atomic:operations": [
{
"op": "update",
"ref": {
"type": "computer",
"id": computer_1.id,
"relationship": "user",
},
"data": {
"type": "user",
"id": user_1.id,
},
},
],
}
response = await client.post("/operations", json=operation_data)
assert response.status_code == status.HTTP_200_OK, response.text
await async_session.refresh(computer_1)
assert computer_1.user_id == user_1.id
@pytest.mark.skip("todo: create relationships resources")
async def test_update_to_one_relationship_clear_atomic(
self,
client: AsyncClient,
async_session: AsyncSession,
user_1: User,
computer_1: Computer,
):
"""
# TODO: relationships resources
# as in doc
https://jsonapi.org/ext/atomic/#auto-id-updating-to-one-relationships
{
"atomic:operations": [{
"op": "update",
"ref": {
"type": "articles",
"id": "13",
"relationship": "author"
},
"data": null
}]
}
:param client:
:param async_session:
:param user_1:
:param computer_1:
:return:
"""
computer_1.user_id = user_1.id
await async_session.commit()
await async_session.refresh(computer_1)
assert computer_1.user_id == user_1.id
operation_data = {
"atomic:operations": [
{
"op": "update",
"ref": {
"type": "computer",
"id": computer_1.id,
"relationship": "author",
},
"data": None,
},
],
}
response = await client.post("/operations", json=operation_data)
assert response.status_code == status.HTTP_200_OK, response.text
await async_session.refresh(computer_1)
assert computer_1.user_id == user_1.id