1import logging
2from typing import List, Tuple
3
4import pytest
5
6logging.basicConfig(
7 level=logging.INFO,
8 format="%(asctime)s test %(levelname)s: %(message)s",
9 datefmt="%Y-%m-%d %H:%M:%S",
10)
11
12logger = logging.getLogger("ambassador")
13
14from ambassador import IR, Cache
15from ambassador.compile import Compile
16
17
18def require_no_errors(ir: IR):
19 assert ir.aconf.errors == {}
20
21
22def require_errors(ir: IR, errors: List[Tuple[str, str]]):
23 flattened_ir_errors: List[str] = []
24
25 for key in ir.aconf.errors.keys():
26 for error in ir.aconf.errors[key]:
27 flattened_ir_errors.append(f"{key}: {error['error']}")
28
29 flattened_wanted_errors: List[str] = [f"{key}: {error}" for key, error in errors]
30
31 assert sorted(flattened_ir_errors) == sorted(flattened_wanted_errors)
32
33
34@pytest.mark.compilertest
35def test_valid_forward_client_cert_details():
36 yaml = """
37---
38apiVersion: getambassador.io/v3alpha1
39kind: Module
40metadata:
41 name: ambassador
42 namespace: default
43spec:
44 config:
45 forward_client_cert_details: SANITIZE_SET
46"""
47
48 cache = Cache(logger)
49 r1 = Compile(logger, yaml, k8s=True)
50 r2 = Compile(logger, yaml, k8s=True, cache=cache)
51
52 require_no_errors(r1["ir"])
53 require_no_errors(r2["ir"])
54
55
56@pytest.mark.compilertest
57def test_invalid_forward_client_cert_details():
58 yaml = """
59---
60apiVersion: getambassador.io/v3alpha1
61kind: Module
62metadata:
63 name: ambassador
64 namespace: default
65spec:
66 config:
67 forward_client_cert_details: SANITIZE_INVALID
68"""
69
70 cache = Cache(logger)
71 r1 = Compile(logger, yaml, k8s=True)
72 r2 = Compile(logger, yaml, k8s=True, cache=cache)
73
74 require_errors(
75 r1["ir"],
76 [
77 (
78 "ambassador.default.1",
79 "'forward_client_cert_details' may not be set to 'SANITIZE_INVALID'; it may only be set to one of: SANITIZE, FORWARD_ONLY, APPEND_FORWARD, SANITIZE_SET, ALWAYS_FORWARD_ONLY",
80 )
81 ],
82 )
83 require_errors(
84 r2["ir"],
85 [
86 (
87 "ambassador.default.1",
88 "'forward_client_cert_details' may not be set to 'SANITIZE_INVALID'; it may only be set to one of: SANITIZE, FORWARD_ONLY, APPEND_FORWARD, SANITIZE_SET, ALWAYS_FORWARD_ONLY",
89 )
90 ],
91 )
92
93
94@pytest.mark.compilertest
95def test_valid_set_current_client_cert_details():
96 yaml = """
97---
98apiVersion: getambassador.io/v3alpha1
99kind: Module
100metadata:
101 name: ambassador
102 namespace: default
103spec:
104 config:
105 set_current_client_cert_details:
106 subject: true
107 dns: true
108"""
109
110 cache = Cache(logger)
111 r1 = Compile(logger, yaml, k8s=True)
112 r2 = Compile(logger, yaml, k8s=True, cache=cache)
113
114 require_no_errors(r1["ir"])
115 require_no_errors(r2["ir"])
116
117
118@pytest.mark.compilertest
119def test_invalid_set_current_client_cert_details_key():
120 yaml = """
121---
122apiVersion: getambassador.io/v3alpha1
123kind: Module
124metadata:
125 name: ambassador
126 namespace: default
127spec:
128 config:
129 set_current_client_cert_details:
130 invalid: true
131"""
132
133 cache = Cache(logger)
134 r1 = Compile(logger, yaml, k8s=True)
135 r2 = Compile(logger, yaml, k8s=True, cache=cache)
136
137 logger.info("R1 IR: %s", r1["ir"].as_json())
138
139 require_errors(
140 r1["ir"],
141 [
142 (
143 "ambassador.default.1",
144 "'set_current_client_cert_details' may not contain key 'invalid'; it may only contain keys: subject, cert, chain, dns, uri",
145 )
146 ],
147 )
148 require_errors(
149 r2["ir"],
150 [
151 (
152 "ambassador.default.1",
153 "'set_current_client_cert_details' may not contain key 'invalid'; it may only contain keys: subject, cert, chain, dns, uri",
154 )
155 ],
156 )
157
158
159@pytest.mark.compilertest
160def test_invalid_set_current_client_cert_details_value():
161 yaml = """
162---
163apiVersion: getambassador.io/v3alpha1
164kind: Module
165metadata:
166 name: ambassador
167 namespace: default
168spec:
169 config:
170 set_current_client_cert_details:
171 subject: invalid
172"""
173
174 cache = Cache(logger)
175 r1 = Compile(logger, yaml, k8s=True)
176 r2 = Compile(logger, yaml, k8s=True, cache=cache)
177
178 require_errors(
179 r1["ir"],
180 [
181 (
182 "ambassador.default.1",
183 "'set_current_client_cert_details' value for key 'subject' may only be 'true' or 'false', not 'invalid'",
184 )
185 ],
186 )
187 require_errors(
188 r2["ir"],
189 [
190 (
191 "ambassador.default.1",
192 "'set_current_client_cert_details' value for key 'subject' may only be 'true' or 'false', not 'invalid'",
193 )
194 ],
195 )
196
197
198@pytest.mark.compilertest
199def test_valid_grpc_stats_all_methods():
200 yaml = """
201---
202apiVersion: getambassador.io/v3alpha1
203kind: Module
204metadata:
205 name: ambassador
206 namespace: default
207spec:
208 config:
209 grpc_stats:
210 all_methods: true
211"""
212
213 cache = Cache(logger)
214 r1 = Compile(logger, yaml, k8s=True)
215 r2 = Compile(logger, yaml, k8s=True, cache=cache)
216
217 require_no_errors(r1["ir"])
218 require_no_errors(r2["ir"])
219
220 ir = r1["ir"].as_dict()
221 stats_filters = [f for f in ir["filters"] if f["name"] == "grpc_stats"]
222 assert len(stats_filters) == 1
223 assert stats_filters[0]["config"] == {
224 "enable_upstream_stats": False,
225 "stats_for_all_methods": True,
226 }
227
228
229@pytest.mark.compilertest
230def test_valid_grpc_stats_services():
231 yaml = """
232---
233apiVersion: getambassador.io/v3alpha1
234kind: Module
235metadata:
236 name: ambassador
237 namespace: default
238spec:
239 config:
240 grpc_stats:
241 services:
242 - name: echo.EchoService
243 method_names: [Echo]
244"""
245
246 cache = Cache(logger)
247 r1 = Compile(logger, yaml, k8s=True)
248 r2 = Compile(logger, yaml, k8s=True, cache=cache)
249
250 require_no_errors(r1["ir"])
251 require_no_errors(r2["ir"])
252
253 ir = r1["ir"].as_dict()
254 stats_filters = [f for f in ir["filters"] if f["name"] == "grpc_stats"]
255 assert len(stats_filters) == 1
256 assert stats_filters[0]["config"] == {
257 "enable_upstream_stats": False,
258 "individual_method_stats_allowlist": {
259 "services": [{"name": "echo.EchoService", "method_names": ["Echo"]}]
260 },
261 }
262
263
264@pytest.mark.compilertest
265def test_valid_grpc_stats_upstream():
266 yaml = """
267---
268apiVersion: getambassador.io/v3alpha1
269kind: Module
270metadata:
271 name: ambassador
272 namespace: default
273spec:
274 config:
275 grpc_stats:
276 upstream_stats: true
277"""
278
279 cache = Cache(logger)
280 r1 = Compile(logger, yaml, k8s=True)
281 r2 = Compile(logger, yaml, k8s=True, cache=cache)
282
283 require_no_errors(r1["ir"])
284 require_no_errors(r2["ir"])
285
286 ir = r1["ir"].as_dict()
287 stats_filters = [f for f in ir["filters"] if f["name"] == "grpc_stats"]
288 assert len(stats_filters) == 1
289 assert stats_filters[0]["config"] == {
290 "enable_upstream_stats": True,
291 "stats_for_all_methods": False,
292 }
293
294
295@pytest.mark.compilertest
296def test_invalid_grpc_stats():
297 yaml = """
298---
299apiVersion: getambassador.io/v3alpha1
300kind: Module
301metadata:
302 name: ambassador
303 namespace: default
304spec:
305 config:
306 grpc_stats:
307"""
308
309 cache = Cache(logger)
310 r1 = Compile(logger, yaml, k8s=True)
311 r2 = Compile(logger, yaml, k8s=True, cache=cache)
312
313 require_no_errors(r1["ir"])
314 require_no_errors(r2["ir"])
315
316 ir = r1["ir"].as_dict()
317 stats_filters = [f for f in ir["filters"] if f["name"] == "grpc_stats"]
318 assert len(stats_filters) == 0
319
320
321@pytest.mark.compilertest
322def test_valid_grpc_stats_empty():
323 yaml = """
324---
325apiVersion: getambassador.io/v3alpha1
326kind: Module
327metadata:
328 name: ambassador
329 namespace: default
330spec:
331 config:
332 grpc_stats: {}
333"""
334
335 cache = Cache(logger)
336 r1 = Compile(logger, yaml, k8s=True)
337 r2 = Compile(logger, yaml, k8s=True, cache=cache)
338
339 require_no_errors(r1["ir"])
340 require_no_errors(r2["ir"])
341
342 ir = r1["ir"].as_dict()
343 stats_filters = [f for f in ir["filters"] if f["name"] == "grpc_stats"]
344 assert len(stats_filters) == 1
345 assert stats_filters[0]["config"] == {
346 "enable_upstream_stats": False,
347 "stats_for_all_methods": False,
348 }
View as plain text