test(client): pin 1.7.0 backend_metrics shape with deterministic structural assertions

hugegraph-python-client/src/tests/api/test_metric.py

@@ -23,6 +23,41 @@
 
 pytestmark = [pytest.mark.integration, pytest.mark.hugegraph]
 
+# Expected server-entry keys for RocksDB backend
+# Captured from a real HugeGraph 1.7.0 server response at /metrics/backend
+EXPECTED_ROCKSDB_SERVER_KEYS = {
+    "mem_unit",
+    "disk_unit",
+    "mem_used",
+    "mem_used_readable",
+    "disk_usage",
+    "disk_usage_readable",
+    "block_cache_usage",
+    "block_cache_pinned_usage",
+    "block_cache_capacity",
+    "estimate_table_readers_mem",
+    "size_all_mem_tables",
+    "cur_size_all_mem_tables",
+    "estimate_live_data_size",
+    "total_sst_files_size",
+    "live_sst_files_size",
+    "estimate_pending_compaction_bytes",
+    "estimate_num_keys",
+    "num_entries_active_mem_table",
+    "num_entries_imm_mem_tables",
+    "num_deletes_active_mem_table",
+    "num_deletes_imm_mem_tables",
+    "num_running_flushes",
+    "mem_table_flush_pending",
+    "num_running_compactions",
+    "compaction_pending",
+    "num_immutable_mem_table",
+    "num_snapshots",
+    "oldest_snapshot_time",
+    "num_live_versions",
+    "current_super_version_number",
+}
+
 
 class TestMetricsManager(unittest.TestCase):
     client = None
@@ -74,10 +109,57 @@ def test_metrics_operations(self):
         self.assertIsInstance(statistics, dict)
 
         backend_metrics = self.metrics.get_backend_metrics()
-        # In HugeGraph 1.7.0+, the backend_metrics structure changed
-        # It's still a dict, but the "hugegraph" key may not exist in the same format
-        self.assertIsInstance(backend_metrics, dict)
-        self.assertTrue(backend_metrics, "backend metrics should not be empty")
-        # Only assert on the "hugegraph" key if it exists (for backward compatibility)
-        if "hugegraph" in backend_metrics:
-            self.assertGreater(len(backend_metrics["hugegraph"]), 1)
+
+        # HugeGraph 1.7.0 backend_metrics shape (top-level key may be "<cluster>-<graph>"):
+        # { "<cluster>-<graph>": { "backend": str, "nodes": int, "cluster_id": str,
+        #                          "servers": { "<server_name>": { <metrics> } } } }
+        self.assertIsInstance(backend_metrics, dict, "backend_metrics should be a dict")
+        self.assertTrue(backend_metrics, "backend_metrics should not be empty")
+
+        # Select the graph entry deterministically using the configured graph name
+        graph_name = self.client.GRAPH
+        graph_key = next(
+            (k for k in backend_metrics if k == graph_name or k.endswith(f"-{graph_name}")),
+            None,
+        )
+        self.assertIsNotNone(
+            graph_key,
+            f"Expected a key matching '{graph_name}' in backend_metrics, got: {list(backend_metrics.keys())}",
+        )
+
+        graph_entry = backend_metrics[graph_key]
+        self.assertIsInstance(graph_entry, dict)
+
+        # Assert required top-level fields in graph entry (backend-agnostic contract)
+        self.assertIn("backend", graph_entry, "Missing 'backend' field")
+        self.assertIn("nodes", graph_entry, "Missing 'nodes' field")
+        self.assertIn("cluster_id", graph_entry, "Missing 'cluster_id' field")
+        self.assertIn("servers", graph_entry, "Missing 'servers' field")
+        self.assertIsInstance(graph_entry["backend"], str)
+        self.assertIsInstance(graph_entry["nodes"], int)
+        self.assertIsInstance(graph_entry["cluster_id"], str)
+        self.assertIsInstance(graph_entry["servers"], dict)
+
+        # Assert every server entry — branch by backend type for portability
+        servers = graph_entry["servers"]
+        self.assertTrue(servers, "servers should not be empty")
+        backend_type = graph_entry["backend"]
+        for server_name, server_entry in servers.items():
+            self.assertIsInstance(
+                server_entry,
+                dict,
+                f"backend_metrics server entry for {server_name} should be a dict",
+            )
+            if backend_type == "rocksdb":
+                # Pin RocksDB-specific keys captured from HugeGraph 1.7.0
+                missing_keys = EXPECTED_ROCKSDB_SERVER_KEYS - set(server_entry.keys())
+                self.assertFalse(
+                    missing_keys,
+                    f"rocksdb server entry for {server_name} missing expected keys: {sorted(missing_keys)}",
+                )
+            else:
+                # Backend-agnostic: assert non-empty dict only
+                self.assertTrue(
+                    server_entry,
+                    f"server entry for {server_name} should not be empty",
+                )