This is an automated email from the ASF dual-hosted git repository.
dmeden pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/trafficserver.git
The following commit(s) were added to refs/heads/master by this push:
new 1f9186e7d8 Test: `traffic_ctl_output_test*`: Add validate functions
for specific JSON fields validation (#12627)
1f9186e7d8 is described below
commit 1f9186e7d8c8efae7432b649c2aeeaf874ec4f5f
Author: Damian Meden <[email protected]>
AuthorDate: Tue Nov 4 10:41:34 2025 +0100
Test: `traffic_ctl_output_test*`: Add validate functions for specific JSON
fields validation (#12627)
* Test: Add validate_json_contains for traffic_ctl output JSON validation
Add validate_json_contains() to validate specific JSON fields without
exact output matches. Updated traffic_ctl_server_output.test.py to use
the new method for server status checks, making tests more maintainable.
Also re-order some functions.
---
.../traffic_ctl/traffic_ctl_server_output.test.py | 13 ++---
.../traffic_ctl/traffic_ctl_test_utils.py | 68 ++++++++++++++++------
2 files changed, 56 insertions(+), 25 deletions(-)
diff --git a/tests/gold_tests/traffic_ctl/traffic_ctl_server_output.test.py
b/tests/gold_tests/traffic_ctl/traffic_ctl_server_output.test.py
index b42e29ebb5..237448d04c 100644
--- a/tests/gold_tests/traffic_ctl/traffic_ctl_server_output.test.py
+++ b/tests/gold_tests/traffic_ctl/traffic_ctl_server_output.test.py
@@ -40,16 +40,13 @@ Test.Summary = 'Basic test for traffic_ctl server command
features.'
traffic_ctl = Make_traffic_ctl(Test, records_yaml)
######
# traffic_ctl server status
-traffic_ctl.server().status().validate_with_text(
- '{"initialized_done": "true", "is_ssl_handshaking_stopped": "false",
"is_draining": "false", "is_event_system_shut_down": "false", "thread_groups":
[{"name": "ET_NET", "count": "``", "started": "true"}, {"name": "ET_TASK",
"count": "2", "started": "true"}, {"name": "ET_UDP", "count": "0", "started":
"false"}]}'
-)
+traffic_ctl.server().status().validate_json_contains(
+ initialized_done='true', is_ssl_handshaking_stopped='false',
is_draining='false', is_event_system_shut_down='false')
# Drain ats so we can check the output.
traffic_ctl.server().drain().exec()
# After the drain, server status should reflect this change.
-traffic_ctl.server().status().validate_with_text(
- '{"initialized_done": "true", "is_ssl_handshaking_stopped": "false",
"is_draining": "true", "is_event_system_shut_down": "false", "thread_groups":
[{"name": "ET_NET", "count": "``", "started": "true"}, {"name": "ET_TASK",
"count": "2", "started": "true"}, {"name": "ET_UDP", "count": "0", "started":
"false"}]}'
-)
+traffic_ctl.server().status().validate_json_contains(initialized_done='true',
is_draining='true')
# Get basic and empty connection tracker info.
traffic_ctl.rpc().invoke(
@@ -58,11 +55,11 @@ traffic_ctl.rpc().invoke(
# default = outbound only
traffic_ctl.rpc().invoke(
handler="get_connection_tracker_info").validate_result_with_text('{"outbound":
{"count": "0", "list": []}}')
-# requets inbound oonly
+# request inbound only
traffic_ctl.rpc().invoke(
handler="get_connection_tracker_info",
params='"table: inbound"').validate_result_with_text('{"inbound":
{"count": "0", "list": []}}')
-# requets outbound only
+# request outbound only
traffic_ctl.rpc().invoke(
handler="get_connection_tracker_info",
params='"table: outbound"').validate_result_with_text('{"outbound":
{"count": "0", "list": []}}')
diff --git a/tests/gold_tests/traffic_ctl/traffic_ctl_test_utils.py
b/tests/gold_tests/traffic_ctl/traffic_ctl_test_utils.py
index a51bb665cc..afd923b74c 100644
--- a/tests/gold_tests/traffic_ctl/traffic_ctl_test_utils.py
+++ b/tests/gold_tests/traffic_ctl/traffic_ctl_test_utils.py
@@ -71,6 +71,57 @@ class Common():
self._finish_callback(self)
return self
+ def validate_with_text(self, text: str):
+ """
+ Validate command output matches expected text exactly.
+
+ Example:
+
traffic_ctl.config().get("proxy.config.product_name").validate_with_text("Apache
Traffic Server")
+ """
+ self._tr.Processes.Default.Streams.stdout = MakeGoldFileWithText(text,
self._dir, self._tn)
+ self._finish_callback(self)
+ return self
+
+ def validate_result_with_text(self, text: str):
+ """
+ Validate RPC result matches expected JSON exactly. Wraps text in
JSON-RPC envelope.
+
+ Example:
+
traffic_ctl.rpc().invoke(handler="get_connection_tracker_info").validate_result_with_text(
+ '{"outbound": {"count": "0", "list": []}}'
+ )
+ """
+ full_text = f'{{\"jsonrpc\": \"2.0\", \"result\": {text}, \"id\":
{"``"}}}'
+ self._tr.Processes.Default.Streams.stdout =
MakeGoldFileWithText(full_text, self._dir, self._tn)
+ self._finish_callback(self)
+ return self
+
+ def validate_json_contains(self, **field_checks):
+ """
+ Validate JSON output contains specific field:value pairs. Only checks
specified fields.
+ Prints detailed error on failure: "FAIL: field_name = actual_value
(expected expected_value)"
+ stream.all.txt will contain the actual output with the failed fields.
+
+ Example:
+ traffic_ctl.server().status().validate_json_contains(
+ initialized_done='true', is_draining='false'
+ )
+ """
+ import json
+ checks_str = ', '.join(f"'{k}': '{v}'" for k, v in
field_checks.items())
+ self._cmd = (
+ f'{self._cmd} | python3 -c "'
+ f"import sys, json; "
+ f"d = json.load(sys.stdin); "
+ f"c = {{{checks_str}}}; "
+ f"failed = [(k, v, str(d.get(k))) for k, v in c.items() if
str(d.get(k)) != v]; "
+ f"[print(f'FAIL: {{k}} = {{actual}} (expected {{expected}})',
file=sys.stderr) "
+ f"for k, expected, actual in failed]; "
+ f"exit(0 if not failed else 1)"
+ f'"')
+ self._finish_callback(self)
+ return self
+
class Config(Common):
"""
@@ -119,10 +170,6 @@ class Config(Common):
self._tr.Processes.Default.Streams.stdout = os.path.join("gold", file)
self.__finish()
- def validate_with_text(self, text: str):
- self._tr.Processes.Default.Streams.stdout = MakeGoldFileWithText(text,
self._dir, self._tn)
- self.__finish()
-
class Server(Common):
"""
@@ -165,10 +212,6 @@ class Server(Common):
"""
self._tr.Processes.Default.Command = self._cmd
- def validate_with_text(self, text: str):
- self._tr.Processes.Default.Streams.stdout = MakeGoldFileWithText(text,
self._dir, self._tn)
- self.__finish()
-
class RPC(Common):
"""
@@ -197,15 +240,6 @@ class RPC(Common):
"""
self._tr.Processes.Default.Command = self._cmd
- def validate_with_text(self, text: str):
- self._tr.Processes.Default.Streams.stdout = MakeGoldFileWithText(text,
self._dir, self._tn)
- self.__finish()
-
- def validate_result_with_text(self, text: str):
- full_text = f'{{\"jsonrpc\": \"2.0\", \"result\": {text}, \"id\":
{"``"}}}'
- self._tr.Processes.Default.Streams.stdout =
MakeGoldFileWithText(full_text, self._dir, self._tn)
- self.__finish()
-
'''
