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 self.results[0].backend
58 assert (
59 self.results[0].backend.name == self.target.path.k8s
60 ), f"r0 wanted {self.target.path.k8s} got {self.results[0].backend.name}"
61 assert self.results[1].backend
62 assert (
63 self.results[1].backend.name == self.target2.path.k8s
64 ), f"r1 wanted {self.target2.path.k8s} got {self.results[1].backend.name}"
65
66
67class HeaderRoutingAuth(ServiceType):
68 skip_variant: ClassVar[bool] = True
69
70 def __init__(self, *args, **kwargs) -> None:
71 # Do this unconditionally, since that's part of the point of this class.
72 kwargs[
73 "service_manifests"
74 ] = """
75---
76kind: Service
77apiVersion: v1
78metadata:
79 name: {self.path.k8s}
80spec:
81 selector:
82 backend: {self.path.k8s}
83 ports:
84 - name: http
85 protocol: TCP
86 port: 80
87 targetPort: 80
88 - name: https
89 protocol: TCP
90 port: 443
91 targetPort: 443
92---
93apiVersion: v1
94kind: Pod
95metadata:
96 name: {self.path.k8s}
97 labels:
98 backend: {self.path.k8s}
99spec:
100 containers:
101 - name: backend
102 image: {images[test-auth]}
103 ports:
104 - containerPort: 80
105 env:
106 - name: BACKEND
107 value: {self.path.k8s}
108"""
109
110 super().__init__(*args, **kwargs)
111
112 def requirements(self):
113 yield ("url", Query("http://%s/ambassador/check/" % self.path.fqdn))
114
115
116class AuthenticationHeaderRouting(AmbassadorTest):
117 target1: ServiceType
118 target2: ServiceType
119 auth: ServiceType
120
121 def init(self):
122 if EDGE_STACK:
123 self.xfail = "XFailing for now, custom AuthServices not supported in Edge Stack"
124 self.target1 = HTTP(name="target1")
125 self.target2 = HTTP(name="target2")
126 self.auth = HeaderRoutingAuth()
127
128 def config(self) -> Generator[Union[str, Tuple[Node, str]], None, None]:
129 # The auth service we're using works like this:
130 #
131 # prefix ENDS WITH /good/ -> 200, include X-Auth-Route -> we should hit target2
132 # prefix ENDS WITH /nohdr/ -> 200, no X-Auth-Route -> we should hit target1
133 # anything else -> 403 -> we should see the 403
134
135 yield self, self.format(
136 """
137---
138apiVersion: getambassador.io/v3alpha1
139kind: AuthService
140name: {self.auth.path.k8s}
141auth_service: "{self.auth.path.fqdn}"
142proto: http
143path_prefix: ""
144timeout_ms: 5000
145
146allowed_authorization_headers:
147- X-Auth-Route
148- Extauth
149"""
150 )
151 yield self.target1, self.format(
152 """
153---
154apiVersion: getambassador.io/v3alpha1
155kind: Mapping
156name: {self.name}-target1
157hostname: "*"
158prefix: /target/
159service: http://{self.target1.path.fqdn}
160"""
161 )
162 yield self.target2, self.format(
163 """
164---
165apiVersion: getambassador.io/v3alpha1
166kind: Mapping
167name: {self.name}-target2
168hostname: "*"
169prefix: /target/
170service: http://{self.target2.path.fqdn}
171headers:
172 X-Auth-Route: Route
173"""
174 )
175
176 def queries(self):
177 # [0]
178 yield Query(self.url("target/"), expected=403)
179
180 # [1]
181 yield Query(self.url("target/good/"), expected=200)
182
183 # [2]
184 yield Query(self.url("target/nohdr/"), expected=200)
185
186 # [3]
187 yield Query(self.url("target/crap/"), expected=403)
188
189 def check(self):
190 # [0] should be a 403 from auth
191 assert self.results[0].backend
192 assert (
193 self.results[0].backend.name == self.auth.path.k8s
194 ), f"r0 wanted {self.auth.path.k8s} got {self.results[0].backend.name}"
195
196 # [1] should go to target2
197 assert self.results[1].backend
198 assert (
199 self.results[1].backend.name == self.target2.path.k8s
200 ), f"r1 wanted {self.target2.path.k8s} got {self.results[1].backend.name}"
201
202 # [2] should go to target1
203 assert self.results[2].backend
204 assert (
205 self.results[2].backend.name == self.target1.path.k8s
206 ), f"r2 wanted {self.target1.path.k8s} got {self.results[2].backend.name}"
207
208 # [3] should be a 403 from auth
209 assert self.results[3].backend
210 assert (
211 self.results[3].backend.name == self.auth.path.k8s
212 ), f"r3 wanted {self.auth.path.k8s} got {self.results[3].backend.name}"
View as plain text