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