|
| 1 | +import json |
1 | 2 | import logging |
| 3 | +from datetime import datetime, timezone |
2 | 4 | from collections import defaultdict |
3 | 5 | from itertools import chain, zip_longest |
4 | 6 | from json import dumps |
5 | | -from typing import Dict, List |
| 7 | +from typing import Dict, List, Optional |
6 | 8 | from uuid import UUID, uuid4 |
7 | 9 |
|
8 | 10 | from fastapi import FastAPI, status |
9 | 11 | from httpx import AsyncClient |
10 | 12 | from pydantic import BaseModel, Field |
11 | 13 | from pytest import fixture, mark, param # noqa PT013 |
| 14 | +from sqlalchemy import select |
12 | 15 | from sqlalchemy.ext.asyncio import AsyncSession |
13 | 16 |
|
14 | 17 | from fastapi_jsonapi.views.view_base import ViewBase |
|
17 | 20 | from tests.misc.utils import fake |
18 | 21 | from tests.models import ( |
19 | 22 | Computer, |
| 23 | + ContainsTimestamp, |
20 | 24 | IdCast, |
21 | 25 | Post, |
22 | 26 | PostComment, |
@@ -1215,6 +1219,96 @@ async def test_create_with_relationship_to_the_same_table(self): |
1215 | 1219 | "meta": None, |
1216 | 1220 | } |
1217 | 1221 |
|
| 1222 | + async def test_create_with_timestamp(self, async_session: AsyncSession): |
| 1223 | + resource_type = "contains_timestamp_model" |
| 1224 | + |
| 1225 | + class ContainsTimestampAttrsSchema(BaseModel): |
| 1226 | + timestamp: datetime |
| 1227 | + |
| 1228 | + app = build_app_custom( |
| 1229 | + model=ContainsTimestamp, |
| 1230 | + schema=ContainsTimestampAttrsSchema, |
| 1231 | + schema_in_post=ContainsTimestampAttrsSchema, |
| 1232 | + schema_in_patch=ContainsTimestampAttrsSchema, |
| 1233 | + resource_type=resource_type, |
| 1234 | + ) |
| 1235 | + |
| 1236 | + create_timestamp = datetime.now(tz=timezone.utc) |
| 1237 | + create_user_body = { |
| 1238 | + "data": { |
| 1239 | + "attributes": { |
| 1240 | + "timestamp": create_timestamp.isoformat(), |
| 1241 | + }, |
| 1242 | + }, |
| 1243 | + } |
| 1244 | + |
| 1245 | + async with AsyncClient(app=app, base_url="http://test") as client: |
| 1246 | + url = app.url_path_for(f"get_{resource_type}_list") |
| 1247 | + res = await client.post(url, json=create_user_body) |
| 1248 | + assert res.status_code == status.HTTP_201_CREATED, res.text |
| 1249 | + response_json = res.json() |
| 1250 | + |
| 1251 | + assert (entity_id := response_json["data"]["id"]) |
| 1252 | + assert response_json == { |
| 1253 | + "meta": None, |
| 1254 | + "jsonapi": {"version": "1.0"}, |
| 1255 | + "data": { |
| 1256 | + "type": "contains_timestamp_model", |
| 1257 | + "attributes": {"timestamp": create_timestamp.isoformat()}, |
| 1258 | + "id": entity_id, |
| 1259 | + }, |
| 1260 | + } |
| 1261 | + |
| 1262 | + stms = select(ContainsTimestamp).where(ContainsTimestamp.id == int(entity_id)) |
| 1263 | + entity_model: Optional[ContainsTimestamp] = (await async_session.execute(stms)).scalar_one_or_none() |
| 1264 | + assert entity_model |
| 1265 | + assert entity_model.timestamp == create_timestamp |
| 1266 | + |
| 1267 | + params = { |
| 1268 | + "filter": json.dumps( |
| 1269 | + [ |
| 1270 | + { |
| 1271 | + "name": "timestamp", |
| 1272 | + "op": "eq", |
| 1273 | + "val": create_timestamp.isoformat(), |
| 1274 | + }, |
| 1275 | + ], |
| 1276 | + ), |
| 1277 | + } |
| 1278 | + |
| 1279 | + # successfully filtered |
| 1280 | + res = await client.get(url, params=params) |
| 1281 | + assert res.status_code == status.HTTP_200_OK, res.text |
| 1282 | + assert res.json() == { |
| 1283 | + "meta": {"count": 1, "totalPages": 1}, |
| 1284 | + "jsonapi": {"version": "1.0"}, |
| 1285 | + "data": [ |
| 1286 | + { |
| 1287 | + "type": "contains_timestamp_model", |
| 1288 | + "attributes": {"timestamp": create_timestamp.isoformat()}, |
| 1289 | + "id": entity_id, |
| 1290 | + }, |
| 1291 | + ], |
| 1292 | + } |
| 1293 | + |
| 1294 | + # check filter really work |
| 1295 | + params = { |
| 1296 | + "filter": [ |
| 1297 | + { |
| 1298 | + "name": "timestamp", |
| 1299 | + "op": "eq", |
| 1300 | + "val": datetime.now(tz=timezone.utc).isoformat(), |
| 1301 | + }, |
| 1302 | + ], |
| 1303 | + } |
| 1304 | + res = await client.get(url, params=params) |
| 1305 | + assert res.status_code == status.HTTP_200_OK, res.text |
| 1306 | + assert res.json() == { |
| 1307 | + "meta": {"count": 0, "totalPages": 1}, |
| 1308 | + "jsonapi": {"version": "1.0"}, |
| 1309 | + "data": [], |
| 1310 | + } |
| 1311 | + |
1218 | 1312 |
|
1219 | 1313 | class TestPatchObjects: |
1220 | 1314 | async def test_patch_object( |
|
0 commit comments