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