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
|
# Copyright 2023 Amazon.com, Inc. or its affiliates. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"). You
# may not use this file except in compliance with the License. A copy of
# the License is located at
#
# http://aws.amazon.com/apache2.0/
#
# or in the "license" file accompanying this file. This file is
# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
# ANY KIND, either express or implied. See the License for the specific
# language governing permissions and limitations under the License.
import gzip
import pytest
from botocore.compress import COMPRESSION_MAPPING
from botocore.config import Config
from tests import ALL_SERVICES, ClientHTTPStubber, patch_load_service_model
from tests.functional.test_useragent import (
get_captured_ua_strings,
parse_registered_feature_ids,
)
FAKE_MODEL = {
"version": "2.0",
"documentation": "",
"metadata": {
"apiVersion": "2020-02-02",
"endpointPrefix": "otherservice",
"protocol": "query",
"serviceFullName": "Other Service",
"serviceId": "Other Service",
"signatureVersion": "v4",
"signingName": "otherservice",
"uid": "otherservice-2020-02-02",
},
"operations": {
"MockOperation": {
"name": "MockOperation",
"http": {"method": "POST", "requestUri": "/"},
"input": {"shape": "MockOperationRequest"},
"documentation": "",
"requestcompression": {
"encodings": ["gzip"],
},
},
},
"shapes": {
"MockOpParamList": {
"type": "list",
"member": {"shape": "MockOpParam"},
},
"MockOpParam": {
"type": "structure",
"members": {"MockOpParam": {"shape": "MockOpParamValue"}},
},
"MockOpParamValue": {
"type": "string",
},
"MockOperationRequest": {
"type": "structure",
"required": ["MockOpParamList"],
"members": {
"MockOpParamList": {
"shape": "MockOpParamList",
"documentation": "",
},
},
},
},
}
FAKE_RULESET = {
"version": "1.0",
"parameters": {},
"rules": [
{
"conditions": [],
"type": "endpoint",
"endpoint": {
"url": "https://foo.bar",
"properties": {},
"headers": {},
},
}
],
}
def _all_compression_operations():
for service_model in ALL_SERVICES:
for operation_name in service_model.operation_names:
operation_model = service_model.operation_model(operation_name)
if operation_model.request_compression is not None:
yield operation_model
@pytest.mark.parametrize("operation_model", _all_compression_operations())
def test_no_unknown_compression_encodings(operation_model):
for encoding in operation_model.request_compression["encodings"]:
assert encoding in COMPRESSION_MAPPING.keys(), (
f"Found unknown compression encoding '{encoding}' "
f"in operation {operation_model.name}."
)
def test_compression(patched_session, monkeypatch):
patch_load_service_model(
patched_session, monkeypatch, FAKE_MODEL, FAKE_RULESET
)
client = patched_session.create_client(
"otherservice",
region_name="us-west-2",
config=Config(request_min_compression_size_bytes=100),
)
with ClientHTTPStubber(client, strict=True) as http_stubber:
http_stubber.add_response(status=200, body=b"<response/>")
params_list = [
{"MockOpParam": f"MockOpParamValue{i}"} for i in range(1, 21)
]
client.mock_operation(MockOpParamList=params_list)
param_template = (
"MockOpParamList.member.{i}.MockOpParam=MockOpParamValue{i}"
)
serialized_params = "&".join(
param_template.format(i=i) for i in range(1, 21)
)
additional_params = "Action=MockOperation&Version=2020-02-02"
serialized_body = f"{additional_params}&{serialized_params}"
actual_body = gzip.decompress(http_stubber.requests[0].body)
assert serialized_body.encode('utf-8') == actual_body
def test_user_agent_has_gzip_feature_id(patched_session, monkeypatch):
patch_load_service_model(
patched_session, monkeypatch, FAKE_MODEL, FAKE_RULESET
)
client = patched_session.create_client(
"otherservice",
region_name="us-west-2",
config=Config(request_min_compression_size_bytes=100),
)
with ClientHTTPStubber(client, strict=True) as http_stubber:
http_stubber.add_response(status=200, body=b"<response/>")
params_list = [
{"MockOpParam": f"MockOpParamValue{i}"} for i in range(1, 21)
]
# The mock operation registers `'GZIP_REQUEST_COMPRESSION': 'L'`
client.mock_operation(MockOpParamList=params_list)
ua_string = get_captured_ua_strings(http_stubber)[0]
feature_list = parse_registered_feature_ids(ua_string)
assert 'L' in feature_list
|
