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.

Author: brbzull0

tests/gold_tests/traffic_ctl/traffic_ctl_server_output.test.py

@@ -40,16 +40,13 @@
 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 @@
 # 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": []}}')

tests/gold_tests/traffic_ctl/traffic_ctl_test_utils.py

@@ -71,6 +71,57 @@ def validate_with_exit_code(self, exit_code: int):
         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 @@ def validate_with_goldfile(self, file: str):
         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 @@ def __finish(self):
         """
         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 @@ def __finish(self):
         """
         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()
-
 
 '''