1from typing import Generator, Literal, Tuple, Union, cast
2
3from abstract_tests import ALSGRPC, HTTP, AmbassadorTest, Node, ServiceType
4from kat.harness import Query
5
6
7class LogServiceTest(AmbassadorTest):
8 target: ServiceType
9 specified_protocol_version: Literal["v2", "v3", "default"]
10 expected_protocol_version: Literal["v3", "invalid"]
11 als: ServiceType
12
13 @classmethod
14 def variants(cls) -> Generator[Node, None, None]:
15 for protocol_version in ["v2", "v3", "default"]:
16 yield cls(protocol_version, name="{self.specified_protocol_version}")
17
18 def init(self, protocol_version: Literal["v3", "default"]):
19 self.target = HTTP()
20 self.specified_protocol_version = protocol_version
21 self.expected_protocol_version = cast(
22 Literal["v3", "invalid"], protocol_version if protocol_version in ["v3"] else "invalid"
23 )
24 self.als = ALSGRPC()
25
26 def config(self) -> Generator[Union[str, Tuple[Node, str]], None, None]:
27 yield self, self.format(
28 """
29---
30apiVersion: getambassador.io/v3alpha1
31kind: LogService
32name: custom-http-logging
33service: {self.als.path.fqdn}
34grpc: true
35driver: http
36driver_config:
37 additional_log_headers:
38 - header_name: "included-on-all"
39 - header_name: "not-included-on-trailer"
40 during_trailer: false
41 - header_name: "not-included on resp-trail"
42 during_trailer: false
43 during_response: false
44 - header_name: "not-anywhere"
45 during_trailer: false
46 during_response: false
47 during_request: false
48flush_interval_time: 1
49flush_interval_byte_size: 1
50"""
51 ) + (
52 ""
53 if self.specified_protocol_version == "default"
54 else f"protocol_version: '{self.specified_protocol_version}'"
55 )
56 yield self, self.format(
57 """
58---
59apiVersion: getambassador.io/v3alpha1
60kind: Mapping
61name: accesslog_target_mapping
62hostname: "*"
63prefix: /target/
64service: {self.target.path.fqdn}
65"""
66 )
67
68 def queries(self):
69 yield Query(f"http://{self.als.path.fqdn}/logs", method="DELETE", phase=1)
70 yield Query(self.url("target/foo"), phase=2)
71 yield Query(self.url("target/bar"), phase=3)
72 yield Query(f"http://{self.als.path.fqdn}/logs", phase=4)
73
74 def check(self):
75 logs = self.results[3].json
76 expkey = f"als{self.expected_protocol_version}-http"
77 for key in ["alsv2-http", "alsv2-tcp", "alsv3-http", "alsv3-tcp"]:
78 if key == expkey:
79 continue
80 assert not logs[key]
81
82 if self.expected_protocol_version == "invalid":
83 assert expkey not in logs
84 return
85
86 assert logs[expkey]
87 assert len(logs[expkey]) == 2
88 assert logs[expkey][0]["request"]["original_path"] == "/target/foo"
89 assert logs[expkey][1]["request"]["original_path"] == "/target/bar"
90
91
92class LogServiceLongServiceNameTest(AmbassadorTest):
93 target: ServiceType
94 als: ServiceType
95
96 def init(self):
97 self.target = HTTP()
98 self.als = ALSGRPC()
99
100 def manifests(self) -> str:
101 return (
102 self.format(
103 """
104---
105kind: Service
106apiVersion: v1
107metadata:
108 name: logservicelongservicename-longnamewithnearly60characters
109spec:
110 selector:
111 backend: {self.als.path.k8s}
112 ports:
113 - name: http
114 protocol: TCP
115 port: 80
116 targetPort: 8080
117 - name: https
118 protocol: TCP
119 port: 443
120 targetPort: 8443
121"""
122 )
123 + super().manifests()
124 )
125
126 def config(self) -> Generator[Union[str, Tuple[Node, str]], None, None]:
127 yield self, self.format(
128 """
129---
130apiVersion: getambassador.io/v3alpha1
131kind: LogService
132name: custom-http-logging
133service: logservicelongservicename-longnamewithnearly60characters
134grpc: true
135protocol_version: "v3"
136driver: http
137driver_config:
138 additional_log_headers:
139 - header_name: "included-on-all"
140 - header_name: "not-included-on-trailer"
141 during_trailer: false
142 - header_name: "not-included on resp-trail"
143 during_trailer: false
144 during_response: false
145 - header_name: "not-anywhere"
146 during_trailer: false
147 during_response: false
148 during_request: false
149flush_interval_time: 1
150flush_interval_byte_size: 1
151 """
152 )
153 yield self, self.format(
154 """
155---
156apiVersion: getambassador.io/v3alpha1
157kind: Mapping
158name: accesslog_target_mapping
159hostname: "*"
160prefix: /target/
161service: {self.target.path.fqdn}
162"""
163 )
164
165 def queries(self):
166 yield Query(f"http://{self.als.path.fqdn}/logs", method="DELETE", phase=1)
167 yield Query(self.url("target/foo"), phase=2)
168 yield Query(self.url("target/bar"), phase=3)
169 yield Query(f"http://{self.als.path.fqdn}/logs", phase=4)
170
171 def check(self):
172 logs = self.results[3].json
173 assert not logs["alsv2-http"]
174 assert not logs["alsv2-tcp"]
175 assert logs["alsv3-http"]
176 assert not logs["alsv3-tcp"]
177
178 assert len(logs["alsv3-http"]) == 2
179 assert logs["alsv3-http"][0]["request"]["original_path"] == "/target/foo"
180 assert logs["alsv3-http"][1]["request"]["original_path"] == "/target/bar"
View as plain text