1from typing import ClassVar, Generator, Tuple, Union
2
3from abstract_tests import HTTP, AmbassadorTest, MappingTest, Node, ServiceType
4from kat.harness import EDGE_STACK, Query, variants
5
6
7class HeaderRoutingTest(MappingTest):
8 parent: AmbassadorTest
9 target: ServiceType
10 target2: ServiceType
11 weight: int
12
13 @classmethod
14 def variants(cls) -> Generator[Node, None, None]:
15 for v in variants(ServiceType):
16 yield cls(v, v.clone("target2"), name="{self.target.name}")
17
18 # XXX This type: ignore is here because we're deliberately overriding the
19 # parent's init to have a different signature... but it's also intimately
20 # (nay, incestuously) related to the variant()'s yield() above, and I really
21 # don't want to deal with that right now. So. We'll deal with it later.
22 def init(self, target: ServiceType, target2: ServiceType): # type: ignore
23 MappingTest.init(self, target)
24 self.target2 = target2
25
26 def config(self) -> Generator[Union[str, Tuple[Node, str]], None, None]:
27 yield self.target, self.format(
28 """
29---
30apiVersion: getambassador.io/v3alpha1
31kind: Mapping
32name: {self.name}-target1
33hostname: "*"
34prefix: /{self.name}/
35service: http://{self.target.path.fqdn}
36"""
37 )
38 yield self.target2, self.format(
39 """
40---
41apiVersion: getambassador.io/v3alpha1
42kind: Mapping
43name: {self.name}-target2
44hostname: "*"
45prefix: /{self.name}/
46service: http://{self.target2.path.fqdn}
47headers:
48 X-Route: target2
49"""
50 )
51
52 def queries(self):
53 yield Query(self.parent.url(self.name + "/"))
54 yield Query(self.parent.url(self.name + "/"), headers={"X-Route": "target2"})
55
56 def check(self):
57 assert (
58 self.results[0].backend.name == self.target.path.k8s
59 ), f"r0 wanted {self.target.path.k8s} got {self.results[0].backend.name}"
60 assert (
61 self.results[1].backend.name == self.target2.path.k8s
62 ), f"r1 wanted {self.target2.path.k8s} got {self.results[1].backend.name}"
63
64
65class HeaderRoutingAuth(ServiceType):
66 skip_variant: ClassVar[bool] = True
67
68 def __init__(self, *args, **kwargs) -> None:
69 # Do this unconditionally, since that's part of the point of this class.
70 kwargs[
71 "service_manifests"
72 ] = """
73---
74kind: Service
75apiVersion: v1
76metadata:
77 name: {self.path.k8s}
78spec:
79 selector:
80 backend: {self.path.k8s}
81 ports:
82 - name: http
83 protocol: TCP
84 port: 80
85 targetPort: 80
86 - name: https
87 protocol: TCP
88 port: 443
89 targetPort: 443
90---
91apiVersion: v1
92kind: Pod
93metadata:
94 name: {self.path.k8s}
95 labels:
96 backend: {self.path.k8s}
97spec:
98 containers:
99 - name: backend
100 image: {images[test-auth]}
101 ports:
102 - containerPort: 80
103 env:
104 - name: BACKEND
105 value: {self.path.k8s}
106"""
107
108 super().__init__(*args, **kwargs)
109
110 def requirements(self):
111 yield ("url", Query("http://%s/ambassador/check/" % self.path.fqdn))
112
113
114class AuthenticationHeaderRouting(AmbassadorTest):
115 target1: ServiceType
116 target2: ServiceType
117 auth: ServiceType
118
119 def init(self):
120 if EDGE_STACK:
121 self.xfail = "XFailing for now, custom AuthServices not supported in Edge Stack"
122 self.target1 = HTTP(name="target1")
123 self.target2 = HTTP(name="target2")
124 self.auth = HeaderRoutingAuth()
125
126 def config(self) -> Generator[Union[str, Tuple[Node, str]], None, None]:
127 # The auth service we're using works like this:
128 #
129 # prefix ENDS WITH /good/ -> 200, include X-Auth-Route -> we should hit target2
130 # prefix ENDS WITH /nohdr/ -> 200, no X-Auth-Route -> we should hit target1
131 # anything else -> 403 -> we should see the 403
132
133 yield self, self.format(
134 """
135---
136apiVersion: getambassador.io/v3alpha1
137kind: AuthService
138name: {self.auth.path.k8s}
139auth_service: "{self.auth.path.fqdn}"
140proto: http
141path_prefix: ""
142timeout_ms: 5000
143
144allowed_authorization_headers:
145- X-Auth-Route
146- Extauth
147"""
148 )
149 yield self.target1, self.format(
150 """
151---
152apiVersion: getambassador.io/v3alpha1
153kind: Mapping
154name: {self.name}-target1
155hostname: "*"
156prefix: /target/
157service: http://{self.target1.path.fqdn}
158"""
159 )
160 yield self.target2, self.format(
161 """
162---
163apiVersion: getambassador.io/v3alpha1
164kind: Mapping
165name: {self.name}-target2
166hostname: "*"
167prefix: /target/
168service: http://{self.target2.path.fqdn}
169headers:
170 X-Auth-Route: Route
171"""
172 )
173
174 def queries(self):
175 # [0]
176 yield Query(self.url("target/"), expected=403)
177
178 # [1]
179 yield Query(self.url("target/good/"), expected=200)
180
181 # [2]
182 yield Query(self.url("target/nohdr/"), expected=200)
183
184 # [3]
185 yield Query(self.url("target/crap/"), expected=403)
186
187 def check(self):
188 # [0] should be a 403 from auth
189 assert (
190 self.results[0].backend.name == self.auth.path.k8s
191 ), f"r0 wanted {self.auth.path.k8s} got {self.results[0].backend.name}"
192
193 # [1] should go to target2
194 assert (
195 self.results[1].backend.name == self.target2.path.k8s
196 ), f"r1 wanted {self.target2.path.k8s} got {self.results[1].backend.name}"
197
198 # [2] should go to target1
199 assert (
200 self.results[2].backend.name == self.target1.path.k8s
201 ), f"r2 wanted {self.target1.path.k8s} got {self.results[2].backend.name}"
202
203 # [3] should be a 403 from auth
204 assert (
205 self.results[3].backend.name == self.auth.path.k8s
206 ), f"r3 wanted {self.auth.path.k8s} got {self.results[3].backend.name}"
View as plain text