1from typing import Generator, Tuple, Union
2
3from abstract_tests import HTTP, AmbassadorTest, Node, ServiceType
4from kat.harness import Query
5
6
7class XRequestIdHeaderPreserveTest(AmbassadorTest):
8 target: ServiceType
9
10 def init(self):
11 self.target = HTTP(name="target")
12
13 def config(self) -> Generator[Union[str, Tuple[Node, str]], None, None]:
14 yield self.target, self.format(
15 """
16---
17apiVersion: getambassador.io/v3alpha1
18kind: Module
19name: ambassador
20config:
21 preserve_external_request_id: true
22---
23apiVersion: getambassador.io/v3alpha1
24kind: Mapping
25name: {self.name}-target
26hostname: "*"
27prefix: /target/
28service: http://{self.target.path.fqdn}
29"""
30 )
31
32 def queries(self):
33 yield Query(self.url("target/"), headers={"x-request-id": "hello"})
34
35 def check(self):
36 assert self.results[0].backend.request.headers["x-request-id"] == ["hello"]
37
38
39class XRequestIdHeaderDefaultTest(AmbassadorTest):
40 target: ServiceType
41
42 def init(self):
43 self.xfail = "Need to figure out passing header through external connections from KAT"
44 self.target = HTTP(name="target")
45
46 def config(self) -> Generator[Union[str, Tuple[Node, str]], None, None]:
47 yield self.target, self.format(
48 """
49---
50apiVersion: getambassador.io/v3alpha1
51kind: Module
52name: ambassador
53
54---
55apiVersion: getambassador.io/v3alpha1
56kind: Mapping
57name: {self.name}-target
58hostname: "*"
59prefix: /target/
60service: http://{self.target.path.fqdn}
61"""
62 )
63
64 def queries(self):
65 yield Query(self.url("target/"), headers={"X-Request-Id": "hello"})
66
67 def check(self):
68 assert self.results[0].backend.request.headers["x-request-id"] != ["hello"]
69
70
71# Sanity test that Envoy headers are present if we do not suppress them
72class EnvoyHeadersTest(AmbassadorTest):
73 target: ServiceType
74
75 def init(self):
76 self.target = HTTP(name="target")
77
78 def config(self) -> Generator[Union[str, Tuple[Node, str]], None, None]:
79 yield self.target, self.format(
80 """
81---
82apiVersion: getambassador.io/v3alpha1
83kind: Mapping
84name: {self.name}-target
85hostname: "*"
86prefix: /target/
87rewrite: /rewrite/
88timeout_ms: 5001
89service: http://{self.target.path.fqdn}
90"""
91 )
92
93 def queries(self):
94 yield Query(self.url("target/"))
95
96 def check(self):
97 print("results[0]=%s" % repr(self.results[0]))
98 headers = self.results[0].backend.request.headers
99
100 # All known Envoy headers should be set. The original path header is
101 # include here because we made sure to include a rewrite in the Mapping.
102 assert headers["x-envoy-expected-rq-timeout-ms"] == ["5001"]
103 assert headers["x-envoy-original-path"] == ["/target/"]
104
105
106# Sanity test that we can suppress Envoy headers when configured
107class SuppressEnvoyHeadersTest(AmbassadorTest):
108 target: ServiceType
109
110 def init(self):
111 self.target = HTTP(name="target")
112
113 def config(self) -> Generator[Union[str, Tuple[Node, str]], None, None]:
114 yield self.target, self.format(
115 """
116---
117apiVersion: getambassador.io/v3alpha1
118kind: Module
119name: ambassador
120config:
121 suppress_envoy_headers: true
122---
123apiVersion: getambassador.io/v3alpha1
124kind: Mapping
125name: {self.name}-target
126hostname: "*"
127prefix: /target/
128rewrite: /rewrite/
129timeout_ms: 5001
130service: http://{self.target.path.fqdn}
131"""
132 )
133
134 def queries(self):
135 yield Query(self.url("target/"))
136
137 def check(self):
138 print("results[0]=%s" % repr(self.results[0]))
139 headers = self.results[0].backend.request.headers
140
141 # No Envoy headers should be set
142 assert "x-envoy-expected-rq-timeout-ms" not in headers
143 assert "x-envoy-original-path" not in headers
View as plain text