1import json
2from typing import ClassVar, Dict, Generator, Sequence, Tuple, Union
3
4import pytest
5
6from abstract_tests import MappingTest, Node, OptionTest
7from kat.harness import Query, Test
8
9# This is the place to add new OptionTests.
10
11
12class AddRequestHeaders(OptionTest):
13
14 parent: Test
15
16 VALUES: ClassVar[Sequence[Dict[str, Dict[str, Union[str, bool]]]]] = [
17 {"foo": {"value": "bar"}},
18 {"moo": {"value": "arf"}},
19 {"zoo": {"append": True, "value": "bar"}},
20 {"xoo": {"append": False, "value": "dwe"}},
21 {"aoo": {"value": "tyu"}},
22 ]
23
24 def config(self) -> Generator[Union[str, Tuple[Node, str]], None, None]:
25 yield "add_request_headers: %s" % json.dumps(self.value)
26
27 def check(self):
28 for r in self.parent.results:
29 for k, v in self.value.items():
30 assert r.backend
31 assert r.backend.request
32 actual = r.backend.request.headers.get(k.lower())
33 if isinstance(v, dict):
34 assert actual == [v["value"]], (actual, [v["value"]])
35 else:
36 assert actual == [v], (actual, [v])
37
38
39class AddResponseHeaders(OptionTest):
40
41 parent: Test
42
43 VALUES: ClassVar[Sequence[Dict[str, Dict[str, Union[str, bool]]]]] = [
44 {"foo": {"value": "bar"}},
45 {"moo": {"value": "arf"}},
46 {"zoo": {"append": True, "value": "bar"}},
47 {"xoo": {"append": False, "value": "dwe"}},
48 {"aoo": {"value": "tyu"}},
49 ]
50
51 def config(self) -> Generator[Union[str, Tuple[Node, str]], None, None]:
52 yield "add_response_headers: %s" % json.dumps(self.value)
53
54 def check(self):
55 for r in self.parent.results:
56 # Why do we end up with capitalized headers anyway??
57 lowercased_headers = {k.lower(): v for k, v in r.headers.items()}
58
59 for k, v in self.value.items():
60 actual = lowercased_headers.get(k.lower())
61 if isinstance(v, dict):
62 assert actual == [v["value"]], "expected %s: %s but got %s" % (
63 k,
64 v["value"],
65 lowercased_headers,
66 )
67 else:
68 assert actual == [v], "expected %s: %s but got %s" % (k, v, lowercased_headers)
69
70
71class UseWebsocket(OptionTest):
72 # TODO: add a check with a websocket client as soon as we have backend support for it
73
74 def config(self) -> Generator[Union[str, Tuple[Node, str]], None, None]:
75 yield "use_websocket: true"
76
77
78class CORS(OptionTest):
79 # isolated = True
80 # debug = True
81
82 # Note that there's also a GlobalCORSTest in t_cors.py.
83
84 parent: MappingTest
85
86 def config(self) -> Generator[Union[str, Tuple[Node, str]], None, None]:
87 yield 'cors: { origins: ["*"] }'
88
89 def queries(self):
90 for q in self.parent.queries():
91 yield Query(q.url) # redundant with parent
92 yield Query(q.url, headers={"Origin": "https://www.test-cors.org"})
93
94 def check(self):
95 # can assert about self.parent.results too
96 assert self.results[0].backend
97 assert self.results[0].backend.name == self.parent.target.path.k8s
98 # Uh. Is it OK that this is case-sensitive?
99 assert "Access-Control-Allow-Origin" not in self.results[0].headers
100
101 assert self.results[1].backend
102 assert self.results[1].backend.name == self.parent.target.path.k8s
103 # Uh. Is it OK that this is case-sensitive?
104 assert self.results[1].headers["Access-Control-Allow-Origin"] == [
105 "https://www.test-cors.org"
106 ]
107
108
109class CaseSensitive(OptionTest):
110
111 parent: MappingTest
112
113 def config(self) -> Generator[Union[str, Tuple[Node, str]], None, None]:
114 yield "case_sensitive: false"
115
116 def queries(self):
117 for q in self.parent.queries():
118 idx = q.url.find("/", q.url.find("://") + 3)
119 upped = q.url[:idx] + q.url[idx:].upper()
120 assert upped != q.url
121 yield Query(upped)
122
123
124class AutoHostRewrite(OptionTest):
125
126 parent: MappingTest
127
128 def config(self) -> Generator[Union[str, Tuple[Node, str]], None, None]:
129 yield "auto_host_rewrite: true"
130
131 def check(self):
132 for r in self.parent.results:
133 assert r.backend
134 assert r.backend.request
135 requested_host_echoed = r.backend.request.host
136 responding_host = r.backend.name
137
138 assert requested_host_echoed == self.parent.target.path.fqdn
139 assert responding_host == self.parent.target.path.k8s
140
141
142class Rewrite(OptionTest):
143
144 parent: MappingTest
145
146 VALUES = ("/foo", "foo")
147
148 def config(self) -> Generator[Union[str, Tuple[Node, str]], None, None]:
149 yield self.format("rewrite: {self.value}")
150
151 def queries(self):
152 if self.value[0] != "/":
153 for q in self.parent.pending:
154 q.xfail = "rewrite option is broken for values not beginning in slash"
155
156 return super(OptionTest, self).queries()
157
158 def check(self):
159 if self.value[0] != "/":
160 pytest.xfail("this is broken")
161
162 for r in self.parent.results:
163 assert r.backend
164 assert r.backend.request
165 assert r.backend.request.url.path == self.value
166
167
168class RemoveResponseHeaders(OptionTest):
169
170 parent: Test
171
172 def config(self) -> Generator[Union[str, Tuple[Node, str]], None, None]:
173 yield "remove_response_headers: [x-envoy-upstream-service-time]"
174
175 def check(self):
176 for r in self.parent.results:
177 assert (
178 r.headers.get("x-envoy-upstream-service-time", None) == None
179 ), "x-envoy-upstream-service-time header was meant to be dropped but wasn't"
View as plain text