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
|
from datetime import datetime
from random import randint
import pytest
from fastapi import FastAPI
from pydantic import BaseModel, Field
from starlette.testclient import TestClient
from maggma.api.resource import PostOnlyResource
from maggma.stores import MemoryStore
class Owner(BaseModel):
name: str = Field(..., title="Owner's name")
age: int = Field(None, title="Owne'r Age")
weight: float = Field(None, title="Owner's weight")
last_updated: datetime = Field(None, title="Last updated date for this record")
owners = (
[Owner(name=f"Person{i}", age=i + 3, weight=100 + i) for i in list(range(10))]
+ [Owner(name="PersonAge9", age=9, weight=float(randint(155, 195)))]
+ [Owner(name="PersonWeight150", age=randint(10, 15), weight=float(150))]
+ [Owner(name="PersonAge20Weight200", age=20, weight=float(200))]
)
total_owners = len(owners)
@pytest.fixture()
def owner_store():
store = MemoryStore("owners", key="name")
store.connect()
store.update([d.dict() for d in owners])
return store
def test_init(owner_store):
resource = PostOnlyResource(store=owner_store, model=Owner)
assert len(resource.router.routes) == 2
def test_msonable(owner_store):
owner_resource = PostOnlyResource(store=owner_store, model=Owner)
endpoint_dict = owner_resource.as_dict()
for k in ["@class", "@module", "store", "model"]:
assert k in endpoint_dict
assert isinstance(endpoint_dict["model"], str)
assert endpoint_dict["model"] == "tests.api.test_post_resource.Owner"
def test_post_to_search(owner_store):
endpoint = PostOnlyResource(owner_store, Owner)
app = FastAPI()
app.include_router(endpoint.router)
client = TestClient(app)
assert client.post("/").status_code == 200
@pytest.mark.xfail()
def test_problem_query_params(owner_store):
endpoint = PostOnlyResource(owner_store, Owner)
app = FastAPI()
app.include_router(endpoint.router)
client = TestClient(app)
client.get("/?param=test").status_code
|
