modified: prometheus_api_client/prometheus_connect.py

	modified:   tests/test_prometheus_connect.py

Author: 袁定胜

prometheus_api_client/prometheus_connect.py

@@ -608,13 +608,12 @@ def get_targets(self, state: str = None, scrape_pool: str = None):
                     response.status_code, response.content)
             )
 
-    def get_target_metadata(self, target: dict[str, str], metric: str = None, limit: int = None):
+    def get_target_metadata(self, target: dict[str, str], metric: str = None):
         """
         Get metadata about metrics from a specific target.
 
         :param target: (dict) A dictionary containing target labels to match against (e.g. {'job': 'prometheus'})
         :param metric: (str) Optional metric name to filter metadata
-        :param limit: (int) Optional maximum number of targets to match
         :returns: (list) A list of metadata entries for matching targets
         :raises:
             (RequestException) Raises an exception in case of a connection error
@@ -631,9 +630,6 @@ def get_target_metadata(self, target: dict[str, str], metric: str = None, limit:
                 ",".join(f'{k}="{v}"' for k, v in target.items()) + "}"
             params['match_target'] = match_target
 
-        if limit:
-            params['limit'] = limit
-
         response = self._session.get(
             "{0}/api/v1/targets/metadata".format(self.url),
             verify=self._session.verify,

tests/test_prometheus_connect.py

@@ -145,7 +145,7 @@ def test_get_scrape_pools(self):  # noqa D102
         self.assertTrue(len(scrape_pools) > 0, "no scrape pools found")
         self.assertIsInstance(scrape_pools[0], str)
 
-    def test_get_targets(self):  # noqa D102
+    def test_get_targets(self):   # PR #295
         targets = self.pc.get_targets()
         self.assertIsInstance(targets, dict)
         self.assertIn('activeTargets', targets)
@@ -160,9 +160,8 @@ def test_get_targets(self):  # noqa D102
         if len(scrape_pools := self.pc.get_scrape_pools()) > 0:
             pool_targets = self.pc.get_targets(scrape_pool=scrape_pools[0])
             self.assertIsInstance(pool_targets, dict)
-            self.assertIn('activeTargets', pool_targets)
 
-    def test_get_target_metadata(self):  # noqa D102
+    def test_get_target_metadata(self):   # PR #295
         # Get a target to test with
         targets = self.pc.get_targets()
         if len(targets['activeTargets']) > 0:
@@ -171,42 +170,42 @@ def test_get_target_metadata(self):  # noqa D102
             }
             metadata = self.pc.get_target_metadata(target)
             self.assertIsInstance(metadata, list)
-            
+
             # Test with metric filter
             if len(metadata) > 0:
                 metric_name = metadata[0]['metric']
-                filtered_metadata = self.pc.get_target_metadata(target, metric=metric_name)
+                filtered_metadata = self.pc.get_target_metadata(
+                    target, metric=metric_name)
                 self.assertIsInstance(filtered_metadata, list)
-                self.assertTrue(all(item['metric'] == metric_name for item in filtered_metadata))
-            
-            # Test with limit
-            limited_metadata = self.pc.get_target_metadata(target, limit=1)
-            self.assertLessEqual(len(limited_metadata), 1)
+                self.assertTrue(
+                    all(item['target']['job'] == target['job'] for item in filtered_metadata))
+
 
-    def test_get_metric_metadata(self):  # noqa D102
+    def test_get_metric_metadata(self):  # PR #295
         metadata = self.pc.get_metric_metadata(metric=None)
         self.assertIsInstance(metadata, list)
         self.assertTrue(len(metadata) > 0, "no metric metadata found")
-        
+
         # Check structure of metadata
         self.assertIn('metric_name', metadata[0])
         self.assertIn('type', metadata[0])
         self.assertIn('help', metadata[0])
         self.assertIn('unit', metadata[0])
-        
+
         # Test with specific metric
         if len(metadata) > 0:
             metric_name = metadata[0]['metric_name']
             filtered_metadata = self.pc.get_metric_metadata(metric=metric_name)
             self.assertIsInstance(filtered_metadata, list)
-            self.assertTrue(all(item['metric_name'] == metric_name for item in filtered_metadata))
-        
+            self.assertTrue(
+                all(item['metric_name'] == metric_name for item in filtered_metadata))
+
         # Test with limit
-        limited_metadata = self.pc.get_metric_metadata(limit=1)
+        limited_metadata = self.pc.get_metric_metadata(metric_name, limit=1)
         self.assertLessEqual(len(limited_metadata), 1)
-        
+
         # Test with limit_per_metric
-        limited_per_metric = self.pc.get_metric_metadata(limit_per_metric=1)
+        limited_per_metric = self.pc.get_metric_metadata(metric_name, limit_per_metric=1)
         self.assertIsInstance(limited_per_metric, list)