File: test_server.py

package info (click to toggle)
python-moto 5.1.18-3
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid
  • size: 116,520 kB
  • sloc: python: 636,725; javascript: 181; makefile: 39; sh: 3
file content (146 lines) | stat: -rw-r--r-- 5,140 bytes parent folder | download | duplicates (2)
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
import json

import moto.server as server

"""
Test the different server responses
"""


def test_list_apis():
    backend = server.create_backend_app("apigateway")
    test_client = backend.test_client()

    res = test_client.get("/restapis")
    assert "item" in json.loads(res.data)


def test_usage_plans_apis():
    backend = server.create_backend_app("apigateway")
    test_client = backend.test_client()

    # List usage plans (expect empty)
    res = test_client.get("/usageplans")
    assert len(json.loads(res.data)["item"]) == 0

    # Create usage plan
    res = test_client.post("/usageplans", data=json.dumps({"name": "test"}))
    created_plan = json.loads(res.data)
    assert created_plan["name"] == "test"

    # List usage plans (expect 1 plan)
    res = test_client.get("/usageplans")
    assert len(json.loads(res.data)["item"]) == 1

    # Get single usage plan
    res = test_client.get(f"/usageplans/{created_plan['id']}")
    fetched_plan = json.loads(res.data)
    assert fetched_plan == created_plan

    # Not existing usage plan
    res = test_client.get("/usageplans/not_existing")
    assert res.status_code == 404

    # Delete usage plan
    res = test_client.delete(f"/usageplans/{created_plan['id']}")
    assert res.data == b"{}"

    # List usage plans (expect empty again)
    res = test_client.get("/usageplans")
    assert len(json.loads(res.data)["item"]) == 0


def test_usage_plans_keys():
    backend = server.create_backend_app("apigateway")
    test_client = backend.test_client()
    usage_plan_id = "test_plan_id"

    # Create API key to be used in tests
    res = test_client.post("/apikeys", data=json.dumps({"name": "test"}))
    created_api_key = json.loads(res.data)

    # List usage plans keys (expect empty)
    res = test_client.get(f"/usageplans/{usage_plan_id}/keys")
    assert len(json.loads(res.data)["item"]) == 0

    # Invalid api key (does not exists at all)
    res = test_client.get(f"/usageplans/{usage_plan_id}/keys/not_existing")
    assert res.status_code == 404

    # not existing usage plan with existing api key
    res = test_client.get(f"/usageplans/not_existing/keys/{created_api_key['id']}")
    assert res.status_code == 404

    # not jet added api key
    res = test_client.get(f"/usageplans/{usage_plan_id}/keys/{created_api_key['id']}")
    assert res.status_code == 404

    # Create usage plan key
    res = test_client.post(
        f"/usageplans/{usage_plan_id}/keys",
        data=json.dumps({"keyId": created_api_key["id"], "keyType": "API_KEY"}),
    )
    created_usage_plan_key = json.loads(res.data)

    # List usage plans keys (expect 1 key)
    res = test_client.get(f"/usageplans/{usage_plan_id}/keys")
    assert len(json.loads(res.data)["item"]) == 1

    # Get single usage plan key
    res = test_client.get(f"/usageplans/{usage_plan_id}/keys/{created_api_key['id']}")
    fetched_plan_key = json.loads(res.data)
    assert fetched_plan_key == created_usage_plan_key

    # Delete usage plan key
    res = test_client.delete(
        f"/usageplans/{usage_plan_id}/keys/{created_api_key['id']}"
    )
    assert res.data == b"{}"

    # List usage plans keys (expect to be empty again)
    res = test_client.get(f"/usageplans/{usage_plan_id}/keys")
    assert len(json.loads(res.data)["item"]) == 0


def test_create_usage_plans_key_non_existent_api_key():
    backend = server.create_backend_app("apigateway")
    test_client = backend.test_client()
    usage_plan_id = "test_plan_id"

    # Create usage plan key with non-existent api key
    res = test_client.post(
        f"/usageplans/{usage_plan_id}/keys",
        data=json.dumps({"keyId": "non-existent", "keyType": "API_KEY"}),
    )
    assert res.status_code == 404


def test_put_integration_response_without_body():
    # Method under test: put_integration_response
    #
    # Moto/Boto3 requires the responseTemplates-parameter to have a value - even if it's an empty dict
    # Botocore <= 1.21.65 does not automatically pass this parameter, so Moto will successfully throw an error if it's not supplied
    # However: As of botocore >= 1.22.0, the responseTemplates is automatically supplied - which means we can no longer test this using boto3
    #
    # This was the equivalent boto3-test:
    # with pytest.raises(ClientError) as ex:
    #     client.put_integration_response(
    #         restApiId=api_id, resourceId=root_id, httpMethod="GET", statusCode="200"
    #     )
    # assert ex.value.response["Error"]["Code"] == "BadRequestException"
    # assert ex.value.response["Error"]["Message"] == "Invalid request input"
    #
    # As a workaround, we can create a PUT-request without body, which will force the error
    # Related: # https://github.com/aws/aws-sdk-js/issues/2588
    #
    backend = server.create_backend_app("apigateway")
    test_client = backend.test_client()

    res = test_client.put(
        "/restapis/f_id/resources/r_id/methods/m_id/integration/responses/200"
    )
    assert res.status_code == 400
    assert json.loads(res.data) == {
        "__type": "BadRequestException",
        "message": "Invalid request input",
    }