IT[test_reconfigure_domains.py]: check structured admin output (#994)

Signed-off-by: Evgenii Malygin <[email protected]>

Author: 678098

src/integration-tests/test_reconfigure_domains.py

@@ -1,4 +1,4 @@
-# Copyright 2024 Bloomberg Finance L.P.
+# Copyright 2025 Bloomberg Finance L.P.
 # SPDX-License-Identifier: Apache-2.0
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
@@ -532,16 +532,16 @@ def run_consumers(max_attempts: int) -> int:
         admin = AdminClient()
         admin.connect(*cluster.admin_endpoint)
 
-        res = admin.send_admin(f"DOMAINS DOMAIN {domain_priority} INFOS")
-        assert '"maxDeliveryAttempts" : 0' in res
+        domain_cfg = admin.get_domain_config(domain_priority)
+        assert domain_cfg["maxDeliveryAttempts"] == 0
 
         cluster.config.domains[
             domain_priority
         ].definition.parameters.max_delivery_attempts = 5
         cluster.reconfigure_domain(domain_priority, succeed=True)
 
-        res = admin.send_admin(f"DOMAINS DOMAIN {domain_priority} INFOS")
-        assert '"maxDeliveryAttempts" : 5' in res
+        domain_cfg = admin.get_domain_config(domain_priority)
+        assert domain_cfg["maxDeliveryAttempts"] == 5
 
         admin.stop()
 

src/python/blazingmq/dev/it/process/admin.py

@@ -1,4 +1,4 @@
-# Copyright 2024 Bloomberg Finance L.P.
+# Copyright 2025 Bloomberg Finance L.P.
 # SPDX-License-Identifier: Apache-2.0
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
@@ -20,7 +20,8 @@
 PURPOSE: Provide a BMQ admin client.
 """
 
-from typing import Union
+import json
+from typing import Union, Dict, Any
 
 from blazingmq.schemas import broker
 from .rawclient import RawClient
@@ -62,3 +63,27 @@ def connect(self, host: str, port: int) -> None:
 
         self._send_raw(self._wrap_control_event(admin_client_identity))
         self._receive_event()
+
+    # Admin APIs
+    def get_domain_config(self, domain_name: str) -> Dict[str, Any]:
+        """
+        Send the "DOMAINS DOMAIN <domain_name> INFOS" admin command, parse the received response and
+        return the domain configuration as a structured dictionary.
+        """
+
+        admin_response: str = self.send_admin(
+            f"ENCODING JSON_COMPACT DOMAINS DOMAIN {domain_name} INFOS"
+        )
+
+        # Top-level domain information has the following structure:
+        # {
+        #     "domainInfo": {
+        #         "name": str,
+        #         "configJson": str,
+        #         ...
+        #     }
+        # }
+        # Need to look up to configJson and parse the string
+        domain_stats = json.loads(admin_response)
+        config_json = domain_stats["domainInfo"]["configJson"]
+        return json.loads(config_json)