1import logging
2
3import pytest
4
5from kat.harness import EDGE_STACK
6from tests.utils import econf_foreach_cluster
7
8logging.basicConfig(
9 level=logging.INFO,
10 format="%(asctime)s test %(levelname)s: %(message)s",
11 datefmt="%Y-%m-%d %H:%M:%S",
12)
13
14logger = logging.getLogger("ambassador")
15# logger.setLevel(logging.DEBUG)
16
17from ambassador import IR, Config, EnvoyConfig
18from ambassador.fetch import ResourceFetcher
19from ambassador.utils import NullSecretHandler
20from tests.utils import default_listener_manifests
21
22
23def _get_ext_auth_config(yaml):
24 for listener in yaml["static_resources"]["listeners"]:
25 for filter_chain in listener["filter_chains"]:
26 for f in filter_chain["filters"]:
27 for http_filter in f["typed_config"]["http_filters"]:
28 if http_filter["name"] == "envoy.filters.http.ext_authz":
29 return http_filter
30 return False
31
32
33def _get_envoy_config(yaml, version="V3"):
34 aconf = Config()
35 fetcher = ResourceFetcher(logger, aconf)
36 fetcher.parse_yaml(default_listener_manifests() + yaml, k8s=True)
37
38 aconf.load_all(fetcher.sorted())
39
40 secret_handler = NullSecretHandler(logger, None, None, "0")
41
42 ir = IR(aconf, file_checker=lambda path: True, secret_handler=secret_handler)
43
44 assert ir
45
46 return EnvoyConfig.generate(ir, version)
47
48
49@pytest.mark.compilertest
50def test_irauth_grpcservice_version_v2():
51 if EDGE_STACK:
52 pytest.xfail("XFailing for now, custom AuthServices not supported in Edge Stack")
53 yaml = """
54---
55apiVersion: getambassador.io/v3alpha1
56kind: AuthService
57metadata:
58 name: mycoolauthservice
59 namespace: default
60spec:
61 auth_service: someservice
62 protocol_version: "v2"
63 proto: grpc
64"""
65 econf = _get_envoy_config(yaml, version="V2")
66
67 conf = econf.as_dict()
68 ext_auth_config = _get_ext_auth_config(conf)
69
70 assert ext_auth_config
71
72 assert (
73 ext_auth_config["typed_config"]["grpc_service"]["envoy_grpc"]["cluster_name"]
74 == "cluster_extauth_someservice_default"
75 )
76
77
78@pytest.mark.compilertest
79def test_irauth_grpcservice_version_v3():
80 yaml = """
81---
82apiVersion: getambassador.io/v3alpha1
83kind: AuthService
84metadata:
85 name: mycoolauthservice
86 namespace: default
87spec:
88 auth_service: someservice
89 protocol_version: "v3"
90 proto: grpc
91"""
92 econf = _get_envoy_config(yaml, version="V3")
93
94 conf = econf.as_dict()
95 ext_auth_config = _get_ext_auth_config(conf)
96
97 assert ext_auth_config
98
99 assert (
100 ext_auth_config["typed_config"]["grpc_service"]["envoy_grpc"]["cluster_name"]
101 == "cluster_extauth_someservice_default"
102 )
103 assert ext_auth_config["typed_config"]["transport_api_version"] == "V3"
104
105
106def test_cluster_fields_v3_config():
107 yaml = """
108---
109apiVersion: getambassador.io/v3alpha1
110kind: AuthService
111metadata:
112 name: mycoolauthservice
113 namespace: default
114spec:
115 auth_service: someservice
116 protocol_version: "v3"
117 proto: grpc
118 stats_name: authservice
119"""
120
121 econf = _get_envoy_config(yaml, version="V3")
122
123 conf = econf.as_dict()
124 ext_auth_config = _get_ext_auth_config(conf)
125
126 cluster_name = "cluster_extauth_someservice_default"
127
128 assert ext_auth_config
129 assert (
130 ext_auth_config["typed_config"]["grpc_service"]["envoy_grpc"]["cluster_name"]
131 == cluster_name
132 )
133
134 def check_fields(cluster):
135 assert cluster["alt_stat_name"] == "authservice"
136
137 econf_foreach_cluster(econf.as_dict(), check_fields, name=cluster_name)
138
139
140def test_cluster_fields_v2_config():
141 yaml = """
142---
143apiVersion: getambassador.io/v3alpha1
144kind: AuthService
145metadata:
146 name: mycoolauthservice
147 namespace: default
148spec:
149 auth_service: someservice
150 protocol_version: "v3"
151 proto: grpc
152 stats_name: authservice
153"""
154
155 econf = _get_envoy_config(yaml, version="V2")
156
157 conf = econf.as_dict()
158 ext_auth_config = _get_ext_auth_config(conf)
159
160 cluster_name = "cluster_extauth_someservice_default"
161
162 assert ext_auth_config
163 assert (
164 ext_auth_config["typed_config"]["grpc_service"]["envoy_grpc"]["cluster_name"]
165 == cluster_name
166 )
167
168 def check_fields(cluster):
169 assert cluster["alt_stat_name"] == "authservice"
170
171 econf_foreach_cluster(econf.as_dict(), check_fields, name=cluster_name)
172
173
174@pytest.mark.compilertest
175def test_irauth_grpcservice_version_default():
176 if EDGE_STACK:
177 pytest.xfail("XFailing for now, custom AuthServices not supported in Edge Stack")
178 yaml = """
179---
180apiVersion: getambassador.io/v3alpha1
181kind: AuthService
182metadata:
183 name: mycoolauthservice
184 namespace: default
185spec:
186 auth_service: someservice
187 proto: grpc
188"""
189 econf = _get_envoy_config(yaml, version="V2")
190
191 conf = econf.as_dict()
192 ext_auth_config = _get_ext_auth_config(conf)
193
194 assert ext_auth_config
195
196 assert (
197 ext_auth_config["typed_config"]["grpc_service"]["envoy_grpc"]["cluster_name"]
198 == "cluster_extauth_someservice_default"
199 )
200
201
202@pytest.mark.compilertest
203def test_irauth_grpcservice_version_default_v3():
204 if EDGE_STACK:
205 pytest.xfail("XFailing for now, custom AuthServices not supported in Edge Stack")
206 yaml = """
207---
208apiVersion: getambassador.io/v3alpha1
209kind: AuthService
210metadata:
211 name: mycoolauthservice
212 namespace: default
213spec:
214 auth_service: someservice
215 proto: grpc
216"""
217 econf = _get_envoy_config(yaml, version="V3")
218
219 conf = econf.as_dict()
220 ext_auth_config = _get_ext_auth_config(conf)
221
222 assert ext_auth_config
223
224 assert (
225 ext_auth_config["typed_config"]["grpc_service"]["envoy_grpc"]["cluster_name"]
226 == "cluster_extauth_someservice_default"
227 )
228 assert ext_auth_config["typed_config"]["transport_api_version"] == "V2"
View as plain text