This repository has been archived by the owner on Aug 2, 2024. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathconftest.py
60 lines (49 loc) · 1.64 KB
/
conftest.py
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
import boto3
import pytest
from moto.dynamodb2 import mock_dynamodb2
from serverless_crud import Manager
from serverless_crud.dynamodb import annotation as db
from serverless_crud.model import BaseModel
@db.Model(
key=db.PrimaryKey(id=db.KeyFieldTypes.HASH),
indexes=(db.GlobalSecondaryIndex("by_user", user=db.KeyFieldTypes.HASH, created=db.KeyFieldTypes.RANGE),),
owner_field="user",
)
class Device(BaseModel):
id: str
created: int
user: str = None
@pytest.fixture(autouse=True, scope="function")
def dynamo():
with mock_dynamodb2():
dynamo = boto3.resource("dynamodb")
dynamo.create_table(
TableName="Device",
KeySchema=[{"AttributeName": "id", "KeyType": "HASH"}],
AttributeDefinitions=[
{"AttributeName": "id", "AttributeType": "S"},
{"AttributeName": "user", "AttributeType": "S"},
],
GlobalSecondaryIndexes=[
{
"IndexName": "by_user",
"KeySchema": [
{"AttributeName": "user", "KeyType": "HASH"},
],
"Projection": {"ProjectionType": "ALL"},
},
],
BillingMode="PAY_PER_REQUEST",
)
table = dynamo.Table("Device")
table.put_item(
Item={"id": "xxx-yyy-zzz", "created": 1234567889, "user": "0b44ac7a-c793-4e4d-9b2d-0634be07598f"},
)
yield dynamo
@pytest.fixture(scope="function")
def app():
api = Manager()
api.rest.registry(Device)
api.appsync.registry(Device)
api.graphql.registry(Device)
return api