1import json
2import os
3import subprocess
4import sys
5import time
6
7import pytest
8
9from abstract_tests import HTTP, AmbassadorTest
10from ambassador.utils import parse_bool
11from kat.harness import Query, is_ingress_class_compatible
12from tests.integration.manifests import namespace_manifest
13from tests.integration.utils import KUBESTATUS_PATH
14
15
16class IngressStatusTest1(AmbassadorTest):
17 status_update = {"loadBalancer": {"ingress": [{"ip": "42.42.42.42"}]}}
18
19 def init(self):
20 self.target = HTTP()
21
22 def manifests(self) -> str:
23 return (
24 """
25---
26apiVersion: networking.k8s.io/v1
27kind: Ingress
28metadata:
29 annotations:
30 kubernetes.io/ingress.class: ambassador
31 getambassador.io/ambassador-id: {self.ambassador_id}
32 name: {self.path.k8s}
33spec:
34 rules:
35 - http:
36 paths:
37 - backend:
38 service:
39 name: {self.target.path.k8s}
40 port:
41 number: 80
42 path: /{self.name}/
43 pathType: Prefix
44"""
45 + super().manifests()
46 )
47
48 def queries(self):
49 if True or sys.platform != "darwin":
50 text = json.dumps(self.status_update)
51
52 update_cmd = [
53 KUBESTATUS_PATH,
54 "Service",
55 "-n",
56 "default",
57 "-f",
58 f"metadata.name={self.path.k8s}",
59 "-u",
60 "/dev/fd/0",
61 ]
62 subprocess.run(update_cmd, input=text.encode("utf-8"), timeout=10)
63 # If you run these tests individually, the time between running kubestatus
64 # and the ingress resource actually getting updated is longer than the
65 # time spent waiting for resources to be ready, so this test will fail (most of the time)
66 time.sleep(1)
67
68 yield Query(self.url(self.name + "/"))
69 yield Query(self.url(f"need-normalization/../{self.name}/"))
70
71 def check(self):
72 if not parse_bool(os.environ.get("AMBASSADOR_PYTEST_INGRESS_TEST", "false")):
73 pytest.xfail("AMBASSADOR_PYTEST_INGRESS_TEST not set, xfailing...")
74
75 if False and sys.platform == "darwin":
76 pytest.xfail("not supported on Darwin")
77
78 for r in self.results:
79 if r.backend:
80 assert r.backend.name == self.target.path.k8s, (
81 r.backend.name,
82 self.target.path.k8s,
83 )
84 assert r.backend.request.headers["x-envoy-original-path"][0] == f"/{self.name}/"
85
86 # check for Ingress IP here
87 ingress_cmd = [
88 "tools/bin/kubectl",
89 "get",
90 "-n",
91 "default",
92 "-o",
93 "json",
94 "ingress",
95 self.path.k8s,
96 ]
97 ingress_run = subprocess.Popen(ingress_cmd, stdout=subprocess.PIPE)
98 ingress_out, _ = ingress_run.communicate()
99 ingress_json = json.loads(ingress_out)
100 assert (
101 ingress_json["status"] == self.status_update
102 ), f"Expected Ingress status to be {self.status_update}, got {ingress_json['status']} instead"
103
104
105class IngressStatusTest2(AmbassadorTest):
106 status_update = {"loadBalancer": {"ingress": [{"ip": "84.84.84.84"}]}}
107
108 def init(self):
109 self.target = HTTP()
110
111 def manifests(self) -> str:
112 return (
113 """
114---
115apiVersion: networking.k8s.io/v1
116kind: Ingress
117metadata:
118 annotations:
119 kubernetes.io/ingress.class: ambassador
120 getambassador.io/ambassador-id: {self.ambassador_id}
121 name: {self.path.k8s}
122spec:
123 rules:
124 - http:
125 paths:
126 - backend:
127 service:
128 name: {self.target.path.k8s}
129 port:
130 number: 80
131 path: /{self.name}/
132 pathType: Prefix
133"""
134 + super().manifests()
135 )
136
137 def queries(self):
138 if True or sys.platform != "darwin":
139 text = json.dumps(self.status_update)
140
141 update_cmd = [
142 KUBESTATUS_PATH,
143 "Service",
144 "-n",
145 "default",
146 "-f",
147 f"metadata.name={self.path.k8s}",
148 "-u",
149 "/dev/fd/0",
150 ]
151 subprocess.run(update_cmd, input=text.encode("utf-8"), timeout=10)
152 # If you run these tests individually, the time between running kubestatus
153 # and the ingress resource actually getting updated is longer than the
154 # time spent waiting for resources to be ready, so this test will fail (most of the time)
155 time.sleep(1)
156
157 yield Query(self.url(self.name + "/"))
158 yield Query(self.url(f"need-normalization/../{self.name}/"))
159
160 def check(self):
161 if not parse_bool(os.environ.get("AMBASSADOR_PYTEST_INGRESS_TEST", "false")):
162 pytest.xfail("AMBASSADOR_PYTEST_INGRESS_TEST not set, xfailing...")
163
164 if False and sys.platform == "darwin":
165 pytest.xfail("not supported on Darwin")
166
167 for r in self.results:
168 if r.backend:
169 assert r.backend.name == self.target.path.k8s, (
170 r.backend.name,
171 self.target.path.k8s,
172 )
173 assert r.backend.request.headers["x-envoy-original-path"][0] == f"/{self.name}/"
174
175 # check for Ingress IP here
176 ingress_cmd = [
177 "tools/bin/kubectl",
178 "get",
179 "-n",
180 "default",
181 "-o",
182 "json",
183 "ingress",
184 self.path.k8s,
185 ]
186 ingress_run = subprocess.Popen(ingress_cmd, stdout=subprocess.PIPE)
187 ingress_out, _ = ingress_run.communicate()
188 ingress_json = json.loads(ingress_out)
189 assert (
190 ingress_json["status"] == self.status_update
191 ), f"Expected Ingress status to be {self.status_update}, got {ingress_json['status']} instead"
192
193
194class IngressStatusTestAcrossNamespaces(AmbassadorTest):
195 status_update = {"loadBalancer": {"ingress": [{"ip": "168.168.168.168"}]}}
196
197 def init(self):
198 self.target = HTTP(namespace="alt-namespace")
199
200 def manifests(self) -> str:
201 return (
202 namespace_manifest("alt-namespace")
203 + """
204---
205apiVersion: networking.k8s.io/v1
206kind: Ingress
207metadata:
208 annotations:
209 kubernetes.io/ingress.class: ambassador
210 getambassador.io/ambassador-id: {self.ambassador_id}
211 name: {self.path.k8s}
212 namespace: alt-namespace
213spec:
214 rules:
215 - http:
216 paths:
217 - backend:
218 service:
219 name: {self.target.path.k8s}
220 port:
221 number: 80
222 path: /{self.name}/
223 pathType: Prefix
224"""
225 + super().manifests()
226 )
227
228 def queries(self):
229 if True or sys.platform != "darwin":
230 text = json.dumps(self.status_update)
231
232 update_cmd = [
233 KUBESTATUS_PATH,
234 "Service",
235 "-n",
236 "default",
237 "-f",
238 f"metadata.name={self.path.k8s}",
239 "-u",
240 "/dev/fd/0",
241 ]
242 subprocess.run(update_cmd, input=text.encode("utf-8"), timeout=10)
243 # If you run these tests individually, the time between running kubestatus
244 # and the ingress resource actually getting updated is longer than the
245 # time spent waiting for resources to be ready, so this test will fail (most of the time)
246 time.sleep(1)
247
248 yield Query(self.url(self.name + "/"))
249 yield Query(self.url(f"need-normalization/../{self.name}/"))
250
251 def check(self):
252 if not parse_bool(os.environ.get("AMBASSADOR_PYTEST_INGRESS_TEST", "false")):
253 pytest.xfail("AMBASSADOR_PYTEST_INGRESS_TEST not set, xfailing...")
254
255 if False and sys.platform == "darwin":
256 pytest.xfail("not supported on Darwin")
257
258 for r in self.results:
259 if r.backend:
260 assert r.backend.name == self.target.path.k8s, (
261 r.backend.name,
262 self.target.path.k8s,
263 )
264 assert r.backend.request.headers["x-envoy-original-path"][0] == f"/{self.name}/"
265
266 # check for Ingress IP here
267 ingress_cmd = [
268 "tools/bin/kubectl",
269 "get",
270 "-o",
271 "json",
272 "ingress",
273 self.path.k8s,
274 "-n",
275 "alt-namespace",
276 ]
277 ingress_run = subprocess.Popen(ingress_cmd, stdout=subprocess.PIPE)
278 ingress_out, _ = ingress_run.communicate()
279 ingress_json = json.loads(ingress_out)
280 assert (
281 ingress_json["status"] == self.status_update
282 ), f"Expected Ingress status to be {self.status_update}, got {ingress_json['status']} instead"
283
284
285class IngressStatusTestWithAnnotations(AmbassadorTest):
286 status_update = {"loadBalancer": {"ingress": [{"ip": "200.200.200.200"}]}}
287
288 def init(self):
289 self.target = HTTP()
290
291 def manifests(self) -> str:
292 return (
293 """
294---
295apiVersion: networking.k8s.io/v1
296kind: Ingress
297metadata:
298 annotations:
299 getambassador.io/config: |
300 ---
301 apiVersion: getambassador.io/v3alpha1
302 kind: Mapping
303 name: {self.name}-nested
304 hostname: "*"
305 prefix: /{self.name}-nested/
306 service: http://{self.target.path.fqdn}
307 ambassador_id: [{self.ambassador_id}]
308 kubernetes.io/ingress.class: ambassador
309 getambassador.io/ambassador-id: {self.ambassador_id}
310 name: {self.path.k8s}
311spec:
312 rules:
313 - http:
314 paths:
315 - backend:
316 service:
317 name: {self.target.path.k8s}
318 port:
319 number: 80
320 path: /{self.name}/
321 pathType: Prefix
322"""
323 + super().manifests()
324 )
325
326 def queries(self):
327 text = json.dumps(self.status_update)
328
329 update_cmd = [
330 KUBESTATUS_PATH,
331 "Service",
332 "-n",
333 "default",
334 "-f",
335 f"metadata.name={self.path.k8s}",
336 "-u",
337 "/dev/fd/0",
338 ]
339 subprocess.run(update_cmd, input=text.encode("utf-8"), timeout=10)
340 # If you run these tests individually, the time between running kubestatus
341 # and the ingress resource actually getting updated is longer than the
342 # time spent waiting for resources to be ready, so this test will fail (most of the time)
343 time.sleep(1)
344
345 yield Query(self.url(self.name + "/"))
346 yield Query(self.url(self.name + "-nested/"))
347 yield Query(self.url(f"need-normalization/../{self.name}/"))
348
349 def check(self):
350 if not parse_bool(os.environ.get("AMBASSADOR_PYTEST_INGRESS_TEST", "false")):
351 pytest.xfail("AMBASSADOR_PYTEST_INGRESS_TEST not set, xfailing...")
352
353 # check for Ingress IP here
354 ingress_cmd = [
355 "tools/bin/kubectl",
356 "get",
357 "-n",
358 "default",
359 "-o",
360 "json",
361 "ingress",
362 self.path.k8s,
363 ]
364 ingress_run = subprocess.Popen(ingress_cmd, stdout=subprocess.PIPE)
365 ingress_out, _ = ingress_run.communicate()
366 ingress_json = json.loads(ingress_out)
367 assert (
368 ingress_json["status"] == self.status_update
369 ), f"Expected Ingress status to be {self.status_update}, got {ingress_json['status']} instead"
370
371
372class SameIngressMultipleNamespaces(AmbassadorTest):
373 status_update = {"loadBalancer": {"ingress": [{"ip": "210.210.210.210"}]}}
374
375 def init(self):
376 self.target = HTTP()
377 self.target1 = HTTP(name="target1", namespace="same-ingress-1")
378 self.target2 = HTTP(name="target2", namespace="same-ingress-2")
379
380 def manifests(self) -> str:
381 return (
382 namespace_manifest("same-ingress-1")
383 + """
384---
385apiVersion: networking.k8s.io/v1
386kind: Ingress
387metadata:
388 annotations:
389 kubernetes.io/ingress.class: ambassador
390 getambassador.io/ambassador-id: {self.ambassador_id}
391 name: {self.path.k8s}
392 namespace: same-ingress-1
393spec:
394 rules:
395 - http:
396 paths:
397 - backend:
398 service:
399 name: {self.target.path.k8s}-target1
400 port:
401 number: 80
402 path: /{self.name}-target1/
403 pathType: Prefix
404"""
405 + namespace_manifest("same-ingress-2")
406 + """
407---
408apiVersion: networking.k8s.io/v1
409kind: Ingress
410metadata:
411 annotations:
412 kubernetes.io/ingress.class: ambassador
413 getambassador.io/ambassador-id: {self.ambassador_id}
414 name: {self.path.k8s}
415 namespace: same-ingress-2
416spec:
417 rules:
418 - http:
419 paths:
420 - backend:
421 service:
422 name: {self.target.path.k8s}-target2
423 port:
424 number: 80
425 path: /{self.name}-target2/
426 pathType: Prefix
427"""
428 + super().manifests()
429 )
430
431 def queries(self):
432 if True or sys.platform != "darwin":
433 text = json.dumps(self.status_update)
434
435 update_cmd = [
436 KUBESTATUS_PATH,
437 "Service",
438 "-n",
439 "default",
440 "-f",
441 f"metadata.name={self.path.k8s}",
442 "-u",
443 "/dev/fd/0",
444 ]
445 subprocess.run(update_cmd, input=text.encode("utf-8"), timeout=10)
446 # If you run these tests individually, the time between running kubestatus
447 # and the ingress resource actually getting updated is longer than the
448 # time spent waiting for resources to be ready, so this test will fail (most of the time)
449 time.sleep(1)
450
451 yield Query(self.url(self.name + "-target1/"))
452 yield Query(self.url(self.name + "-target2/"))
453
454 def check(self):
455 if not parse_bool(os.environ.get("AMBASSADOR_PYTEST_INGRESS_TEST", "false")):
456 pytest.xfail("AMBASSADOR_PYTEST_INGRESS_TEST not set, xfailing...")
457
458 if False and sys.platform == "darwin":
459 pytest.xfail("not supported on Darwin")
460
461 for namespace in ["same-ingress-1", "same-ingress-2"]:
462 # check for Ingress IP here
463 ingress_cmd = [
464 "tools/bin/kubectl",
465 "get",
466 "-n",
467 "default",
468 "-o",
469 "json",
470 "ingress",
471 self.path.k8s,
472 "-n",
473 namespace,
474 ]
475 ingress_run = subprocess.Popen(ingress_cmd, stdout=subprocess.PIPE)
476 ingress_out, _ = ingress_run.communicate()
477 ingress_json = json.loads(ingress_out)
478 assert (
479 ingress_json["status"] == self.status_update
480 ), f"Expected Ingress status to be {self.status_update}, got {ingress_json['status']} instead"
481
482
483class IngressStatusTestWithIngressClass(AmbassadorTest):
484 status_update = {"loadBalancer": {"ingress": [{"ip": "42.42.42.42"}]}}
485
486 def init(self):
487 self.target = HTTP()
488
489 if not is_ingress_class_compatible():
490 self.xfail = "IngressClass is not supported in this cluster"
491
492 def manifests(self) -> str:
493 return (
494 """
495---
496apiVersion: rbac.authorization.k8s.io/v1
497kind: ClusterRole
498metadata:
499 name: {self.path.k8s}-ext
500rules:
501- apiGroups: ["networking.k8s.io"]
502 resources: ["ingressclasses"]
503 verbs: ["get", "list", "watch"]
504---
505apiVersion: rbac.authorization.k8s.io/v1
506kind: ClusterRoleBinding
507metadata:
508 name: {self.path.k8s}-ext
509roleRef:
510 apiGroup: rbac.authorization.k8s.io
511 kind: ClusterRole
512 name: {self.path.k8s}-ext
513subjects:
514- kind: ServiceAccount
515 name: {self.path.k8s}
516 namespace: {self.namespace}
517---
518apiVersion: networking.k8s.io/v1
519kind: IngressClass
520metadata:
521 annotations:
522 getambassador.io/ambassador-id: {self.ambassador_id}
523 name: {self.path.k8s}
524spec:
525 controller: getambassador.io/ingress-controller
526---
527apiVersion: networking.k8s.io/v1
528kind: Ingress
529metadata:
530 annotations:
531 getambassador.io/ambassador-id: {self.ambassador_id}
532 name: {self.path.k8s}
533spec:
534 ingressClassName: {self.path.k8s}
535 rules:
536 - http:
537 paths:
538 - backend:
539 service:
540 name: {self.target.path.k8s}
541 port:
542 number: 80
543 path: /{self.name}/
544 pathType: Prefix
545"""
546 + super().manifests()
547 )
548
549 def queries(self):
550 if True or sys.platform != "darwin":
551 text = json.dumps(self.status_update)
552
553 update_cmd = [
554 KUBESTATUS_PATH,
555 "Service",
556 "-n",
557 "default",
558 "-f",
559 f"metadata.name={self.path.k8s}",
560 "-u",
561 "/dev/fd/0",
562 ]
563 subprocess.run(update_cmd, input=text.encode("utf-8"), timeout=10)
564 # If you run these tests individually, the time between running kubestatus
565 # and the ingress resource actually getting updated is longer than the
566 # time spent waiting for resources to be ready, so this test will fail (most of the time)
567 time.sleep(1)
568
569 yield Query(self.url(self.name + "/"))
570 yield Query(self.url(f"need-normalization/../{self.name}/"))
571
572 def check(self):
573 if not parse_bool(os.environ.get("AMBASSADOR_PYTEST_INGRESS_TEST", "false")):
574 pytest.xfail("AMBASSADOR_PYTEST_INGRESS_TEST not set, xfailing...")
575
576 if False and sys.platform == "darwin":
577 pytest.xfail("not supported on Darwin")
578
579 for r in self.results:
580 if r.backend:
581 assert r.backend.name == self.target.path.k8s, (
582 r.backend.name,
583 self.target.path.k8s,
584 )
585 assert r.backend.request.headers["x-envoy-original-path"][0] == f"/{self.name}/"
586
587 # check for Ingress IP here
588 ingress_cmd = [
589 "tools/bin/kubectl",
590 "get",
591 "-n",
592 "default",
593 "-o",
594 "json",
595 "ingress",
596 self.path.k8s,
597 ]
598 ingress_run = subprocess.Popen(ingress_cmd, stdout=subprocess.PIPE)
599 ingress_out, _ = ingress_run.communicate()
600 ingress_json = json.loads(ingress_out)
601 assert (
602 ingress_json["status"] == self.status_update
603 ), f"Expected Ingress status to be {self.status_update}, got {ingress_json['status']} instead"
View as plain text