1import logging
2from typing import TYPE_CHECKING, Optional
3
4import pytest
5
6from tests.selfsigned import TLSCerts
7from tests.utils import assert_valid_envoy_config
8
9logging.basicConfig(
10 level=logging.INFO,
11 format="%(asctime)s test %(levelname)s: %(message)s",
12 datefmt="%Y-%m-%d %H:%M:%S",
13)
14
15logger = logging.getLogger("ambassador")
16
17from ambassador import IR, Config
18from ambassador.envoy import EnvoyConfig
19from ambassador.fetch import ResourceFetcher
20from ambassador.utils import SecretHandler, SecretInfo
21
22if TYPE_CHECKING:
23 from ambassador.ir.irresource import IRResource # pragma: no cover
24
25
26class MockSecretHandler(SecretHandler):
27 def load_secret(
28 self, resource: "IRResource", secret_name: str, namespace: str
29 ) -> Optional[SecretInfo]:
30 return SecretInfo(
31 "fallback-self-signed-cert",
32 "ambassador",
33 "mocked-fallback-secret",
34 TLSCerts["acook"].pubcert,
35 TLSCerts["acook"].privkey,
36 decode_b64=False,
37 )
38
39
40def get_mirrored_config(ads_config):
41 for l in ads_config.get("static_resources", {}).get("listeners"):
42 for fc in l.get("filter_chains"):
43 for f in fc.get("filters"):
44 for vh in f["typed_config"]["route_config"]["virtual_hosts"]:
45 for r in vh.get("routes"):
46 if r["match"]["prefix"] == "/httpbin/":
47 return r
48 return None
49
50
51@pytest.mark.compilertest
52def test_shadow_v3():
53 aconf = Config()
54
55 yaml = """
56---
57apiVersion: getambassador.io/v3alpha1
58kind: Listener
59metadata:
60 name: ambassador-listener-8080
61 namespace: default
62spec:
63 port: 8080
64 protocol: HTTPS
65 securityModel: XFP
66 hostBinding:
67 namespace:
68 from: ALL
69---
70apiVersion: getambassador.io/v3alpha1
71kind: Mapping
72metadata:
73 name: httpbin-mapping
74 namespace: default
75spec:
76 service: httpbin
77 hostname: "*"
78 prefix: /httpbin/
79---
80apiVersion: getambassador.io/v3alpha1
81kind: Mapping
82metadata:
83 name: httpbin-mapping-shadow
84 namespace: default
85spec:
86 service: httpbin-shadow
87 hostname: "*"
88 prefix: /httpbin/
89 shadow: true
90 weight: 10
91"""
92 fetcher = ResourceFetcher(logger, aconf)
93 fetcher.parse_yaml(yaml, k8s=True)
94
95 aconf.load_all(fetcher.sorted())
96
97 secret_handler = MockSecretHandler(logger, "mockery", "/tmp/ambassador/snapshots", "v1")
98 ir = IR(aconf, file_checker=lambda path: True, secret_handler=secret_handler)
99
100 assert ir
101
102 econf = EnvoyConfig.generate(ir, "V3")
103
104 bootstrap_config, ads_config, _ = econf.split_config()
105 ads_config.pop("@type", None)
106
107 mirrored_config = get_mirrored_config(ads_config)
108 assert "request_mirror_policies" in mirrored_config["route"]
109 assert len(mirrored_config["route"]["request_mirror_policies"]) == 1
110 mirror_policy = mirrored_config["route"]["request_mirror_policies"][0]
111 assert mirror_policy["cluster"] == "cluster_shadow_httpbin_shadow_default"
112 assert mirror_policy["runtime_fraction"]["default_value"]["numerator"] == 10
113 assert mirror_policy["runtime_fraction"]["default_value"]["denominator"] == "HUNDRED"
114 assert_valid_envoy_config(ads_config)
115 assert_valid_envoy_config(bootstrap_config)
116
117
118@pytest.mark.compilertest
119def test_shadow_v2():
120 aconf = Config()
121
122 yaml = """
123---
124apiVersion: getambassador.io/v3alpha1
125kind: Listener
126metadata:
127 name: ambassador-listener-8080
128 namespace: default
129spec:
130 port: 8080
131 protocol: HTTPS
132 securityModel: XFP
133 hostBinding:
134 namespace:
135 from: ALL
136---
137apiVersion: getambassador.io/v3alpha1
138kind: Mapping
139metadata:
140 name: httpbin-mapping
141 namespace: default
142spec:
143 service: httpbin
144 hostname: "*"
145 prefix: /httpbin/
146---
147apiVersion: getambassador.io/v3alpha1
148kind: Mapping
149metadata:
150 name: httpbin-mapping-shadow
151 namespace: default
152spec:
153 service: httpbin-shadow
154 hostname: "*"
155 prefix: /httpbin/
156 shadow: true
157 weight: 10
158"""
159 fetcher = ResourceFetcher(logger, aconf)
160 fetcher.parse_yaml(yaml, k8s=True)
161
162 aconf.load_all(fetcher.sorted())
163
164 secret_handler = MockSecretHandler(logger, "mockery", "/tmp/ambassador/snapshots", "v1")
165 ir = IR(aconf, file_checker=lambda path: True, secret_handler=secret_handler)
166
167 assert ir
168
169 econf = EnvoyConfig.generate(ir, "V2")
170
171 bootstrap_config, ads_config, _ = econf.split_config()
172 ads_config.pop("@type", None)
173
174 mirrored_config = get_mirrored_config(ads_config)
175 assert "request_mirror_policy" in mirrored_config["route"]
176 mirror_policy = mirrored_config["route"]["request_mirror_policy"]
177 assert mirror_policy["cluster"] == "cluster_shadow_httpbin_shadow_default"
178 assert mirror_policy["runtime_fraction"]["default_value"]["numerator"] == 10
179 assert mirror_policy["runtime_fraction"]["default_value"]["denominator"] == "HUNDRED"
180 assert_valid_envoy_config(ads_config, v2=True)
181 assert_valid_envoy_config(bootstrap_config, v2=True)
View as plain text